fix inverted multiply, divide by zero, and not passing msg to channel…

…Emit

src/main/resources/application.conf

@@ -31,9 +31,9 @@ netty {
   threads = 0 # auto
 
   flush {
-    step = 100 # minimum number of channels to flush per interval
-    interval-millis = 1 # interval between flush cycles
+    interval-millis = 1 # interval between flush cycles, set to 0 to disable flush queue
     max-delay-millis = 500 # max flush step targets this threshold if passed
+    step = 100 # minimum number of channels to flush per interval
   }
 }
 

src/main/scala/netty/ActorChannelConnector.scala

@@ -25,7 +25,6 @@ final private class ActorChannelConnector(
   private val monitor  = Monitor.connector.flush
 
   scheduler.scheduleOnce(1 second, () => flush())
-
   scheduler.scheduleWithFixedDelay(1 minute, 1 minute): () =>
     monitor.config.step.update(step.get())
     monitor.config.interval.update(interval.get())
@@ -36,7 +35,7 @@ final private class ActorChannelConnector(
     channel.attr(key.client).set(clientPromise.future)
     val channelEmit: ClientEmit = (msg: ipc.ClientIn) =>
       endpoint.emitCounter.increment()
-      emitToChannel(channel, withFlush = endpoint.alwaysFlush)
+      emitToChannel(channel, withFlush = endpoint.alwaysFlush)(msg)
     clients ! Clients.Control.Start(endpoint.behavior(channelEmit), clientPromise)
     channel.closeFuture.addListener:
       new GenericFutureListener[NettyFuture[Void]]:
@@ -63,7 +62,7 @@ final private class ActorChannelConnector(
 
   private def flush(): Unit =
     val qSize           = flushQ.size
-    val maxDelayFactor  = maxDelay.get().toDouble / interval.get()
+    val maxDelayFactor  = interval.get().toDouble / maxDelay.get().atLeast(1)
     var channelsToFlush = step.get().atLeast((qSize * maxDelayFactor).toInt)
 
     monitor.qSize.record(qSize)

src/main/scala/netty/NettyServer.scala

@@ -16,12 +16,12 @@ final class NettyServer(
     config: Config,
     settings: util.SettingStore
 )(using Executor, Scheduler):
-  private val logger = Logger(getClass)
+  private val logger    = Logger(getClass)
+  private val connector = ActorChannelConnector(clients, config, settings)
 
   def start(): Unit =
 
     logger.info("Start")
-
     val port    = config.getInt("http.port")
     val threads = config.getInt("netty.threads")
     val (parent, workers, channelClass) =
@@ -40,7 +40,7 @@ final class NettyServer(
               pipeline.addLast(HttpServerCodec())
               pipeline.addLast(HttpObjectAggregator(4096))
               pipeline.addLast(RequestHandler(router))
-              pipeline.addLast(ProtocolHandler(ActorChannelConnector(clients, config, settings)))
+              pipeline.addLast(ProtocolHandler(connector))
               pipeline.addLast(FrameHandler())
         )