Prevent outbound join queue from discarding data. Make queue and batch size configurable.

Author: stonesmi

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

@@ -32,7 +32,7 @@ public class BenchmarkHelper {
     private final Clock clock = new DefaultClock();
     private final LifecycleContainer lifecycleContainer = new LifecycleContainer(clock);
     private final MetricsProvider metricsProvider = new MetricsProviderImpl();
-    private final JoinTableProvider joinProvider = JoinTableProviderImpl.apply(clock, lifecycleContainer, metricsProvider);
+    private final JoinTableProvider joinProvider = JoinTableProviderImpl.apply(lifecycleContainer);
 
     public Clock getClock() {
         return clock;

example/main-java/src/main/java/org/finos/vuu/VuuExampleMain.java

@@ -71,6 +71,7 @@ public static void main( String[] args )
                         .withViewPortThreads(4),
                 VuuClientConnectionOptions.apply()
                         .withHeartbeat(),
+                VuuJoinTableProviderOptions.apply(),
                 new scala.collection.mutable.ListBuffer<ViewServerModule>().toList(),
                 new scala.collection.mutable.ListBuffer<Plugin>().toList()
         ).withModule(PriceModule.apply(clock, lifecycle, tableDefContainer))

vuu-java/src/test/java/org/finos/vuu/net/rpc/RpcMethodHandlerTest.java

@@ -7,7 +7,7 @@
 import org.finos.toolbox.time.Clock;
 import org.finos.toolbox.time.DefaultClock;
 import org.finos.vuu.core.table.TableContainer;
-import org.finos.vuu.provider.VuuJoinTableProvider;
+import org.finos.vuu.provider.JoinTableProviderImpl;
 import org.finos.vuu.util.ScalaCollectionConverter;
 import org.junit.Test;
 import scala.Option;
@@ -24,7 +24,7 @@ public void should_register_java_function_as_rpc_in_default_handler() {
         Clock clock = new DefaultClock();
         LifecycleContainer lifecycleContainer = new LifecycleContainer(clock);
         MetricsProvider metricsProvider = new MetricsProviderImpl();
-        TableContainer tableContainer = new TableContainer(new VuuJoinTableProvider(clock, lifecycleContainer, metricsProvider), metricsProvider, clock);
+        TableContainer tableContainer = new TableContainer(JoinTableProviderImpl.apply(lifecycleContainer), metricsProvider, clock);
         final DefaultRpcHandler defaultRpcHandler = new DefaultRpcHandler(tableContainer);
         defaultRpcHandler.registerRpc("helloWorld", rpcService::rpcFunction);
 

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

@@ -6,15 +6,15 @@ import org.finos.toolbox.lifecycle.{LifecycleContainer, LifecycleEnabled}
 import org.finos.toolbox.thread.{LifeCycleRunOncePerThreadExecutorRunner, LifeCycleRunner, WorkItem}
 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.module._
 import org.finos.vuu.core.table.{DataTable, TableContainer}
 import org.finos.vuu.feature.inmem.VuuInMemPlugin
 import org.finos.vuu.net._
 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.rpc.JsonSubTypeRegistry
 import org.finos.vuu.net.ws.WebSocketServer
 import org.finos.vuu.plugin.PluginRegistry
 import org.finos.vuu.provider.{JoinTableProvider, JoinTableProviderImpl, Provider, ProviderContainer}
@@ -42,7 +42,7 @@ class VuuServer(config: VuuServerConfig)(implicit lifecycle: LifecycleContainer,
 
   final val sessionContainer = new ClientSessionContainerImpl()
 
-  final val joinProvider: JoinTableProvider = JoinTableProviderImpl()
+  final val joinProvider: JoinTableProvider = JoinTableProviderImpl(config.joinProvider)
 
   final val tableContainer = new TableContainer(joinProvider)
 

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

@@ -28,7 +28,12 @@ object VuuClientConnectionOptions {
   def apply(): VuuClientConnectionOptions = {
     VuuClientConnectionOptionsImpl(true)
   }
+}
 
+object VuuJoinTableProviderOptions {
+  def apply() : VuuJoinTableProviderOptions = {
+    VuuJoinProviderOptionsImpl.apply(batchSize = 100, maxQueueSize = 20_000)
+  }
 }
 
 trait VuuSSLCipherSuiteOptions {
@@ -77,6 +82,13 @@ trait VuuClientConnectionOptions {
   def withHeartbeatDisabled(): VuuClientConnectionOptions
 }
 
+trait VuuJoinTableProviderOptions {
+  def batchSize: Int
+  def maxQueueSize: Int
+  def withBatchSize(maxQueueDepth: Int): VuuJoinTableProviderOptions
+  def withMaxQueueDepth(maxQueueDepth: Int): VuuJoinTableProviderOptions
+}
+
 case class VuuSecurityOptionsImpl(authenticator: Authenticator, loginTokenValidator: LoginTokenValidator) extends VuuSecurityOptions{
   override def withAuthenticator(authenticator: Authenticator): VuuSecurityOptions = this.copy(authenticator = authenticator)
   override def withLoginValidator(tokenValidator: LoginTokenValidator): VuuSecurityOptions = this.copy(loginTokenValidator = tokenValidator)
@@ -114,11 +126,17 @@ case class VuuSSLCipherSuiteOptionsImpl(ciphers: List[String], protocols: List[S
   override def withProtocols(protocols: List[String]): VuuSSLCipherSuiteOptions = this.copy(protocols = protocols)
 }
 
+case class VuuJoinProviderOptionsImpl(batchSize: Int, maxQueueSize: Int) extends VuuJoinTableProviderOptions {
+  override def withBatchSize(batchSize: Int): VuuJoinTableProviderOptions = this.copy(batchSize = batchSize)
+  override def withMaxQueueDepth(maxQueueSize: Int): VuuJoinTableProviderOptions = this.copy(maxQueueSize = maxQueueSize)
+}
+
 case class VuuServerConfig(httpOptions: VuuHttp2ServerOptions = VuuHttp2ServerOptions(),
                            wsOptions: VuuWebSocketOptions = VuuWebSocketOptions(),
                            security: VuuSecurityOptions = VuuSecurityOptions(),
                            threading: VuuThreadingOptions = VuuThreadingOptions(),
                            clientConnection: VuuClientConnectionOptions = VuuClientConnectionOptions(),
+                           joinProvider: VuuJoinTableProviderOptions = VuuJoinTableProviderOptions(),
                            modules: List[ViewServerModule] = List(),
                            plugins: List[Plugin] = List()) {
   def withModule(module: ViewServerModule): VuuServerConfig = {

vuu/src/main/scala/org/finos/vuu/provider/JoinTableProviderImpl.scala

@@ -1,13 +1,16 @@
 package org.finos.vuu.provider
 
-import org.finos.toolbox.jmx.MetricsProvider
 import org.finos.toolbox.lifecycle.LifecycleContainer
-import org.finos.toolbox.time.Clock
+import org.finos.vuu.core.VuuJoinTableProviderOptions
 
 object JoinTableProviderImpl {
 
-  def apply()(implicit timeProvider: Clock, lifecycle: LifecycleContainer, metrics: MetricsProvider): JoinTableProvider = {
-    new VuuJoinTableProvider()
+  def apply()(implicit lifecycleContainer: LifecycleContainer): JoinTableProvider = {
+    apply(VuuJoinTableProviderOptions())
+  }
+
+  def apply(vuuJoinTableProviderOptions: VuuJoinTableProviderOptions)(implicit lifecycle: LifecycleContainer): JoinTableProvider = {
+    new VuuJoinTableProvider(vuuJoinTableProviderOptions)
   }
 
 }

vuu/src/main/scala/org/finos/vuu/provider/VuuJoinTableProvider.scala

@@ -1,10 +1,9 @@
 package org.finos.vuu.provider
 
 import com.typesafe.scalalogging.StrictLogging
-import org.finos.toolbox.jmx.MetricsProvider
 import org.finos.toolbox.lifecycle.LifecycleContainer
-import org.finos.toolbox.time.Clock
 import org.finos.vuu.api.{JoinTableDef, TableDef}
+import org.finos.vuu.core.VuuJoinTableProviderOptions
 import org.finos.vuu.core.table.{DataTable, JoinTable, JoinTableUpdate, RowWithData}
 import org.finos.vuu.provider.join.{JoinDefToJoinTable, JoinManagerEventDataSink, JoinRelations, RightToLeftKeys}
 
@@ -45,11 +44,11 @@ import java.util.concurrent.{ArrayBlockingQueue, ConcurrentHashMap}
  * },
  * }
  */
-class VuuJoinTableProvider(implicit timeProvider: Clock, lifecycle: LifecycleContainer, metrics: MetricsProvider) extends JoinTableProvider with StrictLogging {
+class VuuJoinTableProvider(options: VuuJoinTableProviderOptions)(implicit lifecycle: LifecycleContainer) extends JoinTableProvider with StrictLogging {
 
   lifecycle(this)
 
-  private val outboundQueue = new ArrayBlockingQueue[JoinTableUpdate](20000)
+  private val outboundQueue = new ArrayBlockingQueue[JoinTableUpdate](options.maxQueueSize)
   private val joinRelations = new JoinRelations()
   private val joinSink = new JoinManagerEventDataSink()
   private val rightToLeftKeys = new RightToLeftKeys()
@@ -121,7 +120,7 @@ class VuuJoinTableProvider(implicit timeProvider: Clock, lifecycle: LifecycleCon
     logger.debug("[JoinTableProvider] Submitting joint table event:" + jtu)
 
     //get the processing off the join thread
-    outboundQueue.offer(jtu)
+    outboundQueue.put(jtu)
   }
 
   def eventToRightKey(joinTableDef: JoinTableDef, tableName: String, ev: util.HashMap[String, Any], rightColumn: String): String = {
@@ -252,7 +251,7 @@ class VuuJoinTableProvider(implicit timeProvider: Clock, lifecycle: LifecycleCon
   }
 
   override def runOnce(): Unit = {
-    val updates = new java.util.ArrayList[JoinTableUpdate](100)
+    val updates = new java.util.ArrayList[JoinTableUpdate](options.batchSize)
 
     outboundQueue.drainTo(updates) match {
 

vuu/src/test/scala/org/finos/vuu/core/row/RowBuilderTest.scala

@@ -5,7 +5,7 @@ import org.finos.toolbox.lifecycle.LifecycleContainer
 import org.finos.toolbox.time.{Clock, TestFriendlyClock}
 import org.finos.vuu.api.TableDef
 import org.finos.vuu.core.table.{Columns, InMemDataTable}
-import org.finos.vuu.provider.VuuJoinTableProvider
+import org.finos.vuu.provider.{JoinTableProvider, JoinTableProviderImpl, VuuJoinTableProvider}
 import org.scalatest.GivenWhenThen
 import org.scalatest.featurespec.AnyFeatureSpec
 import org.scalatest.matchers.should.Matchers
@@ -30,7 +30,7 @@ class RowBuilderTest extends AnyFeatureSpec with Matchers with GivenWhenThen{
       implicit val metrics: MetricsProviderImpl = new MetricsProviderImpl
 
       val tableDef = getTableDef
-      val joinTableProvider: VuuJoinTableProvider = new VuuJoinTableProvider()
+      val joinTableProvider: JoinTableProvider = JoinTableProviderImpl()
 
       val (ricColumn, descColumn, currColumn, exchangeColumn, lotSizeColumn) =
         (
@@ -65,7 +65,7 @@ class RowBuilderTest extends AnyFeatureSpec with Matchers with GivenWhenThen{
       implicit val metrics: MetricsProviderImpl = new MetricsProviderImpl
 
       val tableDef = getTableDef
-      val joinTableProvider: VuuJoinTableProvider = new VuuJoinTableProvider()
+      val joinTableProvider: JoinTableProvider = JoinTableProviderImpl()
 
       val (ricColumn, descColumn, currColumn, exchangeColumn, lotSizeColumn) =
         (

vuu/src/test/scala/org/finos/vuu/core/table/join/JoinManagerTest.scala

@@ -4,7 +4,7 @@ import com.typesafe.scalalogging.StrictLogging
 import org.finos.vuu.api._
 import org.finos.vuu.core.table.join.JoinAsserts._
 import org.finos.vuu.core.table.{Columns, KeyObserver, RowKeyUpdate, TableContainer}
-import org.finos.vuu.provider.{JoinTableProvider, VuuJoinTableProvider}
+import org.finos.vuu.provider.{JoinTableProvider, JoinTableProviderImpl, VuuJoinTableProvider}
 import org.finos.vuu.viewport.ViewPortSetup
 import org.finos.toolbox.jmx.{MetricsProvider, MetricsProviderImpl}
 import org.finos.toolbox.lifecycle.LifecycleContainer
@@ -206,7 +206,7 @@ class JoinManagerTest extends AnyFeatureSpec with Matchers with StrictLogging wi
 
       implicit val lifecycle: LifecycleContainer = new LifecycleContainer
 
-      implicit val joinTableProvider: JoinTableProvider = new VuuJoinTableProvider()
+      implicit val joinTableProvider: JoinTableProvider = JoinTableProviderImpl()
 
       val ordersDef = mkeOrdersDef()
 
@@ -258,7 +258,7 @@ class JoinManagerTest extends AnyFeatureSpec with Matchers with StrictLogging wi
     Scenario("Left Outer Join, Right key update") {
       implicit val lifecycle: LifecycleContainer = new LifecycleContainer
 
-      implicit val joinTableProvider: JoinTableProvider = new VuuJoinTableProvider()
+      implicit val joinTableProvider: JoinTableProvider = JoinTableProviderImpl()
 
       val ordersDef = mkeOrdersDef()
 
@@ -307,7 +307,7 @@ class JoinManagerTest extends AnyFeatureSpec with Matchers with StrictLogging wi
 
       implicit val lifecycle: LifecycleContainer = new LifecycleContainer
 
-      implicit val joinTableProvider: JoinTableProvider = new VuuJoinTableProvider()
+      implicit val joinTableProvider: JoinTableProvider = JoinTableProviderImpl()
 
       val ordersDef = mkeOrders2Def()
       val pricesDef = mkePricesDef()
@@ -359,7 +359,7 @@ class JoinManagerTest extends AnyFeatureSpec with Matchers with StrictLogging wi
 
       implicit val lifecycle: LifecycleContainer = new LifecycleContainer
 
-      implicit val joinTableProvider: JoinTableProvider = new VuuJoinTableProvider()
+      implicit val joinTableProvider: JoinTableProvider = JoinTableProviderImpl()
 
       val ordersDef = mkeOrdersDef()
 
@@ -415,7 +415,7 @@ class JoinManagerTest extends AnyFeatureSpec with Matchers with StrictLogging wi
     Scenario("Left Outer Join, Delete Right Record") {
       implicit val lifecycle: LifecycleContainer = new LifecycleContainer
 
-      implicit val joinTableProvider : JoinTableProvider = new VuuJoinTableProvider()
+      implicit val joinTableProvider : JoinTableProvider = JoinTableProviderImpl()
 
       val ordersDef = mkeOrdersDef()
 
@@ -465,7 +465,7 @@ class JoinManagerTest extends AnyFeatureSpec with Matchers with StrictLogging wi
     Scenario("Left Outer Join of Joins") {
       implicit val lifecycle: LifecycleContainer = new LifecycleContainer
 
-      implicit val joinTableProvider: VuuJoinTableProvider = new VuuJoinTableProvider()
+      implicit val joinTableProvider: JoinTableProvider = JoinTableProviderImpl()
 
       val ordersDef = mkeOrdersDef()
       val pricesDef = mkePricesDef()

vuu/src/test/scala/org/finos/vuu/net/rpc/DefaultRpcHandlerTest.scala

@@ -5,7 +5,7 @@ import org.finos.toolbox.lifecycle.LifecycleContainer
 import org.finos.toolbox.time.{Clock, DefaultClock}
 import org.finos.vuu.core.table.TableContainer
 import org.finos.vuu.net.{ClientSessionId, RequestContext}
-import org.finos.vuu.provider.VuuJoinTableProvider
+import org.finos.vuu.provider.JoinTableProviderImpl
 import org.scalatest.BeforeAndAfterEach
 import org.scalatest.featurespec.AnyFeatureSpec
 import org.scalatest.matchers.should.Matchers
@@ -18,7 +18,7 @@ class DefaultRpcHandlerTest extends AnyFeatureSpec with Matchers with BeforeAndA
     implicit val clock: Clock = new DefaultClock
     implicit val lifecycleContainer: LifecycleContainer = new LifecycleContainer
     implicit val metricsProvider: MetricsProvider = new MetricsProviderImpl
-    implicit val tableContainer: TableContainer = new TableContainer(new VuuJoinTableProvider)
+    implicit val tableContainer: TableContainer = new TableContainer(JoinTableProviderImpl())
     handler = new DefaultRpcHandler
   }