added config options for cached event messages, shared by all cahced event systems

src/main/resources/application.conf

@@ -575,6 +575,17 @@ network {
     # before it is dropped. Can be used as a watchdog for hung server sessions.
     outbound-grace-time = 1 minute
   }
+
+  event-caching {
+    # The minimum amount of time a cached message has to wait before being dispatched. (ms)
+    # Ideal circumstances are when message traffic is light.
+    flush-cache-delay = 15
+    # The absolute maximum amount of time a cached message has to wait before being dispatched. (ms)
+    flush-cache-max-delay = 50
+    # The numeric difference between light traffic and heavy traffic.
+    # In this case, the event system's traffic density and the message caching delay are numerically synonymous.
+    message-traffic-threshold = 12
+  }
 }
 
 development {

src/main/scala/net/psforever/actors/session/SessionActor.scala

@@ -392,6 +392,7 @@ class SessionActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], con
       case unknownStamp =>
         log.error(s"received a message from an unknown event system - reply: $envelope, stamp: $unknownStamp")
     }
+    println(s"event-system-rtt: ${System.currentTimeMillis() - envelope.time} ms")
   }
 
   private def handleEnvelopeWithResponseHandler(

src/main/scala/net/psforever/services/base/GenericEventServiceWithCacheAndSupport.scala

@@ -7,12 +7,13 @@ import net.psforever.services.Service
 import net.psforever.services.base.envelope.{GenericMessageEnvelope, GenericResponseEnvelope, MessageEnvelope, MessageTransformationBehavior}
 import net.psforever.services.base.message.EventMessage
 import net.psforever.types.PlanetSideGUID
+import net.psforever.util.Config
 
 import scala.collection.concurrent.{Map => CMap}
 import scala.jdk.CollectionConverters._
 import java.util.concurrent.ConcurrentHashMap
 import scala.concurrent.ExecutionContext.Implicits.global
-import scala.concurrent.duration.DurationInt
+import scala.concurrent.duration.DurationLong
 
 /*
 Adapted from the rating limiting code in PSForever fork https://github.com/Pinapse/giant with permission
@@ -97,8 +98,11 @@ class GenericEventServiceWithCacheAndSupport
   stamp: EventSystemStamp,
   eventSupportServices: List[EventServiceSupport]
 ) extends GenericEventServiceWithSupport(stamp, eventSupportServices) {
-  private val flushCacheWait: Long = 50L //milliseconds
-  private var hasCachedMessages: Boolean = false
+  private val flushCacheDelay: Long = Config.app.network.eventCaching.flushCacheDelay
+  private val flushCacheMaxDelay: Long = Config.app.network.eventCaching.flushCacheMaxDelay
+  private var hasCachedMessages: Int = 0
+  private var lastCachedMessages: Int = 0
+  private val messageThreshold: Long = Config.app.network.eventCaching.messageTrafficThreshold
   private var nextTimeToFlushCache: Long = 0L
   private var emergencyFlush: Cancellable = Default.Cancellable
 
@@ -110,6 +114,14 @@ class GenericEventServiceWithCacheAndSupport
     super.postStop()
   }
 
+  private def adjustedFlushDelay(): Long = {
+    // flushCacheWait <= t <= emergencyFlushMaxDelay
+    math.min(
+      flushCacheDelay + math.max(0, lastCachedMessages - messageThreshold),
+      flushCacheMaxDelay
+    )
+  }
+
   /**
    * If there were previously no messages in the cache,
    * prepare to flush the cache after the intended interval passes,
@@ -117,10 +129,11 @@ class GenericEventServiceWithCacheAndSupport
    * or if a safety timer expires and the cache is flushed in precaution.
    */
   private def tryRetimeFlushCache(): Unit = {
-    if (!hasCachedMessages) {
-      hasCachedMessages = true
-      nextTimeToFlushCache = System.currentTimeMillis() + flushCacheWait
-      emergencyFlush = context.system.scheduler.scheduleOnce(55 milliseconds, self, FlushCachedMessages)
+    hasCachedMessages += 1
+    if (hasCachedMessages == 1) {
+      val flushDelay = adjustedFlushDelay()
+      nextTimeToFlushCache = System.currentTimeMillis() + flushDelay
+      emergencyFlush = context.system.scheduler.scheduleOnce(delay = (1.25f * flushDelay).toLong milliseconds, self, FlushCachedMessages)
     }
   }
 
@@ -145,7 +158,7 @@ class GenericEventServiceWithCacheAndSupport
    * flush the cache messages to the normal event system bus.
    */
   private def tryFlushCache(): Boolean = {
-    val willFlush = hasCachedMessages && nextTimeToFlushCache < System.currentTimeMillis()
+    val willFlush = hasCachedMessages > 0 && nextTimeToFlushCache < System.currentTimeMillis()
     if (willFlush) {
       flushCache()
     }
@@ -165,7 +178,8 @@ class GenericEventServiceWithCacheAndSupport
         map.clear()
       }
     }
-    hasCachedMessages = false
+    lastCachedMessages = hasCachedMessages
+    hasCachedMessages = 0
     emergencyFlush.cancel()
     emergencyFlush = Default.Cancellable
   }

src/main/scala/net/psforever/services/base/envelope/AllEnvelopes.scala

@@ -12,4 +12,6 @@ trait AllEnvelopes {
   def channel: String
   /** specific subscriber endpoint to be excluded (the subscriber should filter themselves upon receipt) */
   def filter: PlanetSideGUID
+
+  val time: Long = System.currentTimeMillis()
 }

src/main/scala/net/psforever/util/Config.scala

@@ -131,7 +131,8 @@ case class AntiCheatConfig(
 
 case class NetworkConfig(
     session: SessionConfig,
-    middleware: MiddlewareConfig
+    middleware: MiddlewareConfig,
+    eventCaching: CachedMessagesConfig
 )
 
 case class MiddlewareConfig(
@@ -147,6 +148,12 @@ case class SessionConfig(
     outboundGraceTime: FiniteDuration
 )
 
+case class CachedMessagesConfig(
+    flushCacheDelay: Long, //milliseconds
+    flushCacheMaxDelay: Long, //milliseconds
+    messageTrafficThreshold: Long
+)
+
 case class GameConfig(
     instantAction: InstantActionConfig,
     amenityAutorepairRate: Float,