Freeze and unfreeze view port

benchmark/src/main/java/org/finos/vuu/benchmark/BenchmarkHelper.java

@@ -46,7 +46,7 @@ public InMemDataTable buildBigTable(int size) {
                 Indices.apply(toScala(List.of(Index.apply("exchange")))),
                 toScalaSeq(List.of("ric"))
         );
-        var table = new InMemDataTable(pricesDef, joinProvider, metricsProvider);
+        var table = new InMemDataTable(pricesDef, joinProvider, metricsProvider, clock);
 
         for (int i = 0; i <= size; i++) {
             var ric = "TST-" + i;

plugin/virtualized-table-plugin/src/main/scala/org/finos/vuu/plugin/virtualized/VirtualizedTablePlugin.scala

@@ -15,7 +15,7 @@ object VirtualizedTablePlugin extends DefaultPlugin {
   final val callableFactory = new VirtualizedViewPortCallableFactory
 
   override def tableFactory(implicit metrics: MetricsProvider): TableFactory = (tableDef: TableDef, tableContainer: TableContainer, joinTableProvider: JoinTableProvider) => {
-    val table = new InMemDataTable(tableDef, joinTableProvider)
+    val table = new InMemDataTable(tableDef, joinTableProvider)(metrics, tableContainer.timeProvider)
     tableContainer.addTable(table)
     table
   }

vuu/src/main/scala/org/finos/vuu/api/TableDef.scala

@@ -2,6 +2,7 @@ package org.finos.vuu.api
 
 import org.finos.vuu.core.auths.RowPermissionChecker
 import org.finos.vuu.core.module.ViewServerModule
+import org.finos.vuu.core.table.DefaultColumnNames.{CreatedTimeColumnName, LastUpdatedTimeColumnName}
 import org.finos.vuu.core.table._
 import org.finos.vuu.feature.inmem.VuuInMemPluginLocator
 import org.finos.vuu.viewport.ViewPort
@@ -117,12 +118,16 @@ class SessionTableDef(name: String,
 
 class TableDef(val name: String,
                val keyField: String,
-               val columns: Array[Column],
+               val customColumns: Array[Column],
                val joinFields: Seq[String],
                val autosubscribe: Boolean = false,
                val links: VisualLinks = VisualLinks(),
                val indices: Indices) extends VuuInMemPluginLocator {
 
+  private val createdTimeColumn: SimpleColumn = SimpleColumn(CreatedTimeColumnName, customColumns.length, DataType.fromString("long"))
+  private val updatedTimeColumn: SimpleColumn = SimpleColumn(LastUpdatedTimeColumnName, customColumns.length + 1, DataType.fromString("long"))
+  val columns: Array[Column] = customColumns ++ Array(createdTimeColumn, updatedTimeColumn)
+
   private var module: ViewServerModule = null;
   private var permissionFunc: (ViewPort, TableContainer) => RowPermissionChecker = null
 

vuu/src/main/scala/org/finos/vuu/client/ClientHelperFns.scala

@@ -66,6 +66,14 @@ object ClientHelperFns {
     vsClient.send(JsonViewServerMessage(requestId, sessionId, token, user, DisableViewPortRequest(vpId)))
   }
 
+  def freezeViewPort(sessionId: String, token: String, user: String, requestId: String, vpId: String)(implicit vsClient: ViewServerClient): Unit = {
+    vsClient.send(JsonViewServerMessage(requestId, sessionId, token, user, FreezeViewPortRequest(vpId)))
+  }
+
+  def unfreezeViewPort(sessionId: String, token: String, user: String, requestId: String, vpId: String)(implicit vsClient: ViewServerClient): Unit = {
+    vsClient.send(JsonViewServerMessage(requestId, sessionId, token, user, UnfreezeViewPortRequest(vpId)))
+  }
+
   def closeTreeNodeAsync(sessionId: String, token: String, user: String, requestId: String, vpId: String, treeKey: String)(implicit vsClient: ViewServerClient): Unit = {
     vsClient.send(JsonViewServerMessage(requestId, sessionId, token, user, CloseTreeNodeRequest(vpId, treeKey)))
   }
@@ -79,7 +87,7 @@ object ClientHelperFns {
   }
 
   def rpcTableUpdate(sessionId: String, token: String, user: String, table: ViewPortTable, key: String, data: Map[String, Any])(implicit vsClient: ViewServerClient): Unit = {
-    vsClient.send(JsonViewServerMessage(RequestId.oneNew().toString, sessionId, token, user, RpcUpdate(table, key, data)))
+    vsClient.send(JsonViewServerMessage(RequestId.oneNew(), sessionId, token, user, RpcUpdate(table, key, data)))
   }
 
   def getViewPortMenusAsync(sessionId: String, token: String, user: String, vpId: String)(implicit vsClient: ViewServerClient): Unit = {

vuu/src/main/scala/org/finos/vuu/core/CoreServerApiHandler.scala

@@ -170,6 +170,28 @@ class CoreServerApiHandler(val viewPortContainer: ViewPortContainer,
     }
   }
 
+  override def process(msg: FreezeViewPortRequest)(ctx: RequestContext): Option[ViewServerMessage] = {
+    Try(viewPortContainer.freezeViewPort(msg.viewPortId)) match {
+      case Success(_) =>
+        logger.debug("View port froze")
+        vsMsg(FreezeViewPortSuccess(msg.viewPortId))(ctx)
+      case Failure(e) =>
+        logger.warn("Failed to freeze viewport", e)
+        vsMsg(FreezeViewPortReject(msg.viewPortId, e.toString))(ctx)
+    }
+  }
+
+  override def process(msg: UnfreezeViewPortRequest)(ctx: RequestContext): Option[ViewServerMessage] = {
+    Try(viewPortContainer.unfreezeViewPort(msg.viewPortId)) match {
+      case Success(_) =>
+        logger.debug("View port unfroze")
+        vsMsg(UnfreezeViewPortSuccess(msg.viewPortId))(ctx)
+      case Failure(e) =>
+        logger.warn("Failed to unfreeze viewport", e)
+        vsMsg(UnfreezeViewPortReject(msg.viewPortId, e.toString))(ctx)
+    }
+  }
+
   def vsMsg(body: MessageBody)(ctx: RequestContext): Option[JsonViewServerMessage] = {
     Some(VsMsg(ctx.requestId, ctx.session.sessionId, ctx.token, ctx.session.user, body))
   }

vuu/src/main/scala/org/finos/vuu/core/VuuServer.scala

@@ -8,15 +8,15 @@ import org.finos.toolbox.time.Clock
 import org.finos.vuu.api.{JoinTableDef, TableDef, ViewPortDef}
 import org.finos.vuu.core.module.{ModuleContainer, RealizedViewServerModule, StaticServedResource, TableDefContainer, ViewServerModule}
 import org.finos.vuu.core.table.{DataTable, TableContainer}
-import org.finos.vuu.feature.inmem.{VuuInMemPlugin, VuuInMemPluginType}
+import org.finos.vuu.feature.inmem.VuuInMemPlugin
 import org.finos.vuu.net._
-import org.finos.vuu.net.flowcontrol.{DefaultFlowController, FlowController, FlowControllerFactory, NoHeartbeatFlowController}
+import org.finos.vuu.net.flowcontrol.FlowControllerFactory
 import org.finos.vuu.net.http.{Http2Server, VuuHttp2Server}
 import org.finos.vuu.net.json.{CoreJsonSerializationMixin, JsonVsSerializer, Serializer}
 import org.finos.vuu.net.rest.RestService
 import org.finos.vuu.net.rpc.{JsonSubTypeRegistry, RpcHandler}
 import org.finos.vuu.net.ws.WebSocketServer
-import org.finos.vuu.plugin.{Plugin, PluginRegistry}
+import org.finos.vuu.plugin.PluginRegistry
 import org.finos.vuu.provider.{JoinTableProvider, JoinTableProviderImpl, Provider, ProviderContainer}
 import org.finos.vuu.viewport._
 

vuu/src/main/scala/org/finos/vuu/core/sort/FilterAndSort.scala

@@ -1,14 +1,14 @@
 package org.finos.vuu.core.sort
 
 import com.typesafe.scalalogging.StrictLogging
-import org.finos.vuu.core.filter.{Filter, FilterClause, NoFilter}
-import org.finos.vuu.core.index._
-import org.finos.vuu.core.table.{Column, DataType, EmptyTablePrimaryKeys, TablePrimaryKeys, ViewPortColumnCreator}
-import org.finos.vuu.viewport.{RowSource, ViewPortColumns, ViewPortVisualLink}
 import org.finos.toolbox.collection.array.ImmutableArray
 import org.finos.vuu.core.auths.RowPermissionChecker
+import org.finos.vuu.core.filter.{Filter, FilterClause}
+import org.finos.vuu.core.index._
 import org.finos.vuu.core.table.column.{Error, Success}
+import org.finos.vuu.core.table.{Column, DataType, DefaultColumnNames, EmptyTablePrimaryKeys, TablePrimaryKeys, ViewPortColumnCreator}
 import org.finos.vuu.feature.inmem.InMemTablePrimaryKeys
+import org.finos.vuu.viewport.{RowSource, ViewPortColumns, ViewPortVisualLink}
 
 case class VisualLinkedFilter(viewPortVisualLink: ViewPortVisualLink) extends Filter with StrictLogging {
 
@@ -78,7 +78,8 @@ case class RowPermissionFilter(checker: RowPermissionChecker) extends Filter wit
   override def dofilter(source: RowSource, primaryKeys: TablePrimaryKeys, vpColumns: ViewPortColumns): TablePrimaryKeys = {
     val filtered = primaryKeys.filter(key => {
       try {
-      checker.canSeeRow(source.pullRow(key, vpColumns))
+        // calling source.pullRow(key) rather than source.pullRow(key, vpColumns) because user might remove the columns we need from view port
+        checker.canSeeRow(source.pullRow(key))
       } catch {
         case e: Exception =>
           logger.error(s"Error while checking row permission for keys $primaryKeys with checker $checker", e)
@@ -90,6 +91,40 @@ case class RowPermissionFilter(checker: RowPermissionChecker) extends Filter wit
   }
 }
 
+case class FrozenTimeFilter(frozenTime: Long) extends Filter with StrictLogging {
+  override def dofilter(source: RowSource, primaryKeys: TablePrimaryKeys, vpColumns: ViewPortColumns): TablePrimaryKeys = {
+    val filtered = primaryKeys.filter(key => {
+      try {
+        val vuuCreatedTimestamp = source.pullRow(key).get(DefaultColumnNames.CreatedTimeColumnName).asInstanceOf[Long]
+        vuuCreatedTimestamp < frozenTime
+      } catch {
+        case e: Exception =>
+          logger.error(s"Error while checking frozen time for keys $primaryKeys with frozen time $frozenTime", e)
+          false
+      }
+    }).toArray
+
+    InMemTablePrimaryKeys(ImmutableArray.from[String](filtered))
+  }
+}
+
+case class RowPermissionAndFrozenTimeFilter(checker: RowPermissionChecker, frozenTime: Long) extends Filter with StrictLogging {
+  override def dofilter(source: RowSource, primaryKeys: TablePrimaryKeys, vpColumns: ViewPortColumns): TablePrimaryKeys = {
+    val filtered = primaryKeys.filter(key => {
+      try {
+        val rowData = source.pullRow(key)
+        rowData.get(DefaultColumnNames.CreatedTimeColumnName).asInstanceOf[Long] < frozenTime && checker.canSeeRow(rowData)
+      } catch {
+        case e: Exception =>
+          logger.error(s"Error while checking row permission and view port frozen time for keys $primaryKeys with checker $checker and frozen time $frozenTime", e)
+          false
+      }
+    }).toArray
+
+    InMemTablePrimaryKeys(ImmutableArray.from[String](filtered))
+  }
+}
+
 case class TwoStepCompoundFilter(first: Filter, second: Filter) extends Filter with StrictLogging {
   override def dofilter(source: RowSource, primaryKeys: TablePrimaryKeys, vpColumns: ViewPortColumns): TablePrimaryKeys = {
 
@@ -118,7 +153,7 @@ case class AntlrBasedFilter(clause: FilterClause) extends Filter with StrictLogg
 
 
 trait FilterAndSort {
-  def filterAndSort(source: RowSource, primaryKeys: TablePrimaryKeys, vpColumns: ViewPortColumns, permission: Option[RowPermissionChecker]): TablePrimaryKeys
+  def filterAndSort(source: RowSource, primaryKeys: TablePrimaryKeys, vpColumns: ViewPortColumns, permission: Option[RowPermissionChecker], viewPortFrozenTime: Option[Long]): TablePrimaryKeys
 
   def filter: Filter
 
@@ -127,15 +162,15 @@ trait FilterAndSort {
 
 case class UserDefinedFilterAndSort(filter: Filter, sort: Sort) extends FilterAndSort with StrictLogging {
 
-  override def filterAndSort(source: RowSource, primaryKeys: TablePrimaryKeys, vpColumns: ViewPortColumns, checkerOption: Option[RowPermissionChecker]): TablePrimaryKeys = {
-    if(primaryKeys == null || primaryKeys.length == 0) {
+  override def filterAndSort(source: RowSource, primaryKeys: TablePrimaryKeys, vpColumns: ViewPortColumns, checkerOption: Option[RowPermissionChecker], viewPortFrozenTime: Option[Long]): TablePrimaryKeys = {
+    if (primaryKeys == null || primaryKeys.length == 0) {
       // nothing to filter or sort
       return primaryKeys
     }
 
     try {
-      val realizedFilter = checkerOption match {
-        case Some(checker) => TwoStepCompoundFilter(RowPermissionFilter(checker), filter)
+      val realizedFilter = createDefaultFilter(checkerOption, viewPortFrozenTime) match {
+        case Some(defaultFilter) => TwoStepCompoundFilter(defaultFilter, filter)
         case None => filter
       }
 
@@ -152,6 +187,25 @@ case class UserDefinedFilterAndSort(filter: Filter, sort: Sort) extends FilterAn
     }
   }
 
+  private def createDefaultFilter(checkerOption: Option[RowPermissionChecker], viewPortFrozenTime: Option[Long]): Option[Filter] = {
+    checkerOption match {
+      case Some(checker) =>
+        viewPortFrozenTime match {
+          case Some(frozenTime) =>
+            Some(RowPermissionAndFrozenTimeFilter(checker, frozenTime))
+          case None =>
+            Some(RowPermissionFilter(checker))
+        }
+      case None =>
+        viewPortFrozenTime match {
+          case Some(t) =>
+            Some(FrozenTimeFilter(t))
+          case None =>
+            None
+        }
+    }
+  }
+
 }
 
 

vuu/src/main/scala/org/finos/vuu/core/table/AutoSubscribeTable.scala

@@ -1,13 +1,13 @@
 package org.finos.vuu.core.table
 
 import com.typesafe.scalalogging.StrictLogging
-import org.finos.vuu.api.TableDef
-import org.finos.vuu.provider.JoinTableProvider
 import io.vertx.core.impl.ConcurrentHashSet
 import org.finos.toolbox.jmx.MetricsProvider
-import org.finos.vuu.core.row.RowBuilder
+import org.finos.toolbox.time.Clock
+import org.finos.vuu.api.TableDef
+import org.finos.vuu.provider.JoinTableProvider
 
-class AutoSubscribeTable(tableDef: TableDef, joinProvider: JoinTableProvider)(implicit override val metrics: MetricsProvider) extends InMemDataTable(tableDef, joinProvider) with StrictLogging {
+class AutoSubscribeTable(tableDef: TableDef, joinProvider: JoinTableProvider)(implicit override val metrics: MetricsProvider, timeProvider: Clock) extends InMemDataTable(tableDef, joinProvider) with StrictLogging {
 
   private val onTrySubscribe = metrics.counter(plusName("trySubscribe.count"))
   private val totalSubscribe = metrics.counter(plusName("total.count"))

vuu/src/main/scala/org/finos/vuu/core/table/DefaultColumnNames.scala

@@ -0,0 +1,6 @@
+package org.finos.vuu.core.table
+
+object DefaultColumnNames {
+  val CreatedTimeColumnName: String = "vuuCreatedTimestamp"
+  val LastUpdatedTimeColumnName: String = "vuuUpdatedTimestamp"
+}

vuu/src/main/scala/org/finos/vuu/core/table/InMemDataTable.scala

@@ -8,6 +8,7 @@ import org.finos.vuu.viewport.{RowProcessor, RowSource, ViewPortColumns}
 import org.finos.toolbox.collection.array.ImmutableArray
 import org.finos.toolbox.jmx.MetricsProvider
 import org.finos.toolbox.text.AsciiUtil
+import org.finos.toolbox.time.Clock
 import org.finos.vuu.core.row.{InMemMapRowBuilder, RowBuilder}
 import org.finos.vuu.feature.inmem.InMemTablePrimaryKeys
 
@@ -94,8 +95,7 @@ trait DataTable extends KeyedObservable[RowKeyUpdate] with RowSource {
   }
 }
 
-case class RowKeyUpdate(key: String, source: RowSource, isDelete: Boolean = false)
-{
+case class RowKeyUpdate(key: String, source: RowSource, isDelete: Boolean = false) {
   override def toString: String = s"RowKeyUpdate($key, ${source.name})"
 }
 
@@ -166,7 +166,7 @@ object EmptyRowData extends RowData {
 }
 
 
-case class InMemDataTableData(data: ConcurrentHashMap[String, RowData], private val primaryKeyValuesInternal: TablePrimaryKeys) extends TableData {
+case class InMemDataTableData(data: ConcurrentHashMap[String, RowData], private val primaryKeyValuesInternal: TablePrimaryKeys)(implicit timeProvider: Clock) extends TableData {
 
   def primaryKeyValues: TablePrimaryKeys = this.primaryKeyValuesInternal
 
@@ -184,13 +184,16 @@ case class InMemDataTableData(data: ConcurrentHashMap[String, RowData], private
   def update(key: String, update: RowData): TableData = {
 
     val table = data.synchronized {
-
+      val now = timeProvider.now()
       data.getOrDefault(key, EmptyRowData) match {
         case row: RowWithData =>
           val mergedData = merge(update, row)
+          mergedData.set(DefaultColumnNames.LastUpdatedTimeColumnName, now)
           data.put(key, mergedData)
           InMemDataTableData(data, primaryKeyValues)
         case EmptyRowData =>
+          update.set(DefaultColumnNames.CreatedTimeColumnName, now)
+          update.set(DefaultColumnNames.LastUpdatedTimeColumnName, now)
           data.put(key, update)
           InMemDataTableData(data, primaryKeyValues + key)
       }
@@ -220,7 +223,7 @@ case class InMemDataTableData(data: ConcurrentHashMap[String, RowData], private
 }
 
 
-class InMemDataTable(val tableDef: TableDef, val joinProvider: JoinTableProvider)(implicit val metrics: MetricsProvider) extends DataTable with KeyedObservableHelper[RowKeyUpdate] with StrictLogging {
+class InMemDataTable(val tableDef: TableDef, val joinProvider: JoinTableProvider)(implicit val metrics: MetricsProvider, timeProvider: Clock) extends DataTable with KeyedObservableHelper[RowKeyUpdate] with StrictLogging {
 
   private final val indices = tableDef.indices.indices
     .map(index => tableDef.columnForName(index.column))
@@ -339,7 +342,7 @@ class InMemDataTable(val tableDef: TableDef, val joinProvider: JoinTableProvider
 
   }
 
-   private def sendColumnToProcessor(key: String, column: Column, value: Any, rowProcessor: RowProcessor): Unit = {
+  private def sendColumnToProcessor(key: String, column: Column, value: Any, rowProcessor: RowProcessor): Unit = {
     rowProcessor.processColumn(column, value)
   }
 
@@ -462,6 +465,7 @@ class InMemDataTable(val tableDef: TableDef, val joinProvider: JoinTableProvider
     }
   }
 
+  // TODO: #1647 remove the timeStamp here
   def processUpdate(rowKey: String, rowData: RowData, timeStamp: Long): Unit = {
 
     onUpdateMeter.mark()

vuu/src/main/scala/org/finos/vuu/core/table/InMemSessionDataTable.scala

@@ -10,11 +10,11 @@ class InMemSessionDataTable private (val clientSessionId: ClientSessionId,
                                      sessionTableDef: SessionTableDef,
                                      joinTableProvider: JoinTableProvider,
                                      final val creationTimestamp: Long)
-                                    (implicit metrics: MetricsProvider) extends InMemDataTable(sessionTableDef, joinTableProvider) with SessionTable {
+                                    (implicit metrics: MetricsProvider, timeProvider: Clock) extends InMemDataTable(sessionTableDef, joinTableProvider) with SessionTable {
 
   def this(clientSessionId: ClientSessionId, sessionTableDef: SessionTableDef, joinTableProvider: JoinTableProvider)
-          (implicit metrics: MetricsProvider, clock: Clock) = {
-    this(clientSessionId, sessionTableDef, joinTableProvider, creationTimestamp = clock.now())
+          (implicit metrics: MetricsProvider, timeProvider: Clock) = {
+    this(clientSessionId, sessionTableDef, joinTableProvider, creationTimestamp = timeProvider.now())
   }
 
   override def name: String = s"session:$clientSessionId/simple-" + sessionTableDef.name + "_" + creationTimestamp.toString

vuu/src/main/scala/org/finos/vuu/core/table/TableContainer.scala

@@ -25,7 +25,7 @@ trait TableContainerMBean {
   def getSubscribedKeys(name: String): String
 }
 
-class TableContainer(val joinTableProvider: JoinTableProvider)(implicit val metrics: MetricsProvider, clock: Clock) extends JmxAble with TableContainerMBean with StrictLogging {
+class TableContainer(val joinTableProvider: JoinTableProvider)(implicit val metrics: MetricsProvider, val timeProvider: Clock) extends JmxAble with TableContainerMBean with StrictLogging {
 
   private val tables = new ConcurrentHashMap[String, DataTable]()
 

vuu/src/main/scala/org/finos/vuu/core/table/TableData.scala

@@ -1,7 +1,5 @@
 package org.finos.vuu.core.table
 
-import org.finos.toolbox.collection.array.ImmutableArray
-
 trait TableData {
   def dataByKey(key: String): RowData
   def update(key: String, update: RowData): TableData

vuu/src/main/scala/org/finos/vuu/feature/TableFactory.scala

@@ -1,6 +1,5 @@
 package org.finos.vuu.feature
 
-import org.finos.toolbox.jmx.MetricsProvider
 import org.finos.vuu.api.TableDef
 import org.finos.vuu.core.table.{DataTable, TableContainer}
 import org.finos.vuu.provider.JoinTableProvider

vuu/src/main/scala/org/finos/vuu/feature/inmem/VuuInMemPlugin.scala

@@ -42,7 +42,7 @@ class VuuInMemPlugin extends DefaultPlugin {
   override def filterFactory: FilterFactory = ???
   override def sortFactory: SortFactory = ???
   override def tableFactory(implicit metrics: MetricsProvider): TableFactory = (tableDef: TableDef, tableContainer: TableContainer, joinTableProvider: JoinTableProvider) => {
-    val table = new InMemDataTable(tableDef, joinTableProvider)
+    val table = new InMemDataTable(tableDef, joinTableProvider)(metrics, tableContainer.timeProvider)
     tableContainer.addTable(table)
     table
   }

vuu/src/main/scala/org/finos/vuu/net/ClientConnectionCreator.scala

@@ -145,6 +145,8 @@ class DefaultMessageHandler(val channel: Channel,
       case req: RemoveViewPortRequest => serverApi.process(req)(ctx)
       case req: EnableViewPortRequest => serverApi.process(req)(ctx)
       case req: DisableViewPortRequest => serverApi.process(req)(ctx)
+      case req: FreezeViewPortRequest => serverApi.process(req)(ctx)
+      case req: UnfreezeViewPortRequest => serverApi.process(req)(ctx)
       case req: GetViewPortMenusRequest => serverApi.process(req)(ctx)
       case req: ViewPortMenuSelectionRpcCall => serverApi.process(req)(ctx)
       case req: ViewPortMenuRowRpcCall => serverApi.process(req)(ctx)