From 9ba1b3048ed29fc73cd65820299872e5da15b53e Mon Sep 17 00:00:00 2001
From: Fate-JH <Jason_DiDonato@yahoo.com>
Date: Mon, 26 Jan 2026 16:49:50 -0500
Subject: [PATCH] adapted but did not implement the event bus rate limiting
 mechanisms from Chinese PSF fork

---
 .../services/base/GenericEventBus.scala       |   4 +-
 .../services/base/GenericEventService.scala   |   1 +
 .../services/base/GenericGuidEventBus.scala   | 106 ++++++++++++++++++
 3 files changed, 110 insertions(+), 1 deletion(-)
 create mode 100644 src/main/scala/net/psforever/services/base/GenericGuidEventBus.scala

diff --git a/src/main/scala/net/psforever/services/base/GenericEventBus.scala b/src/main/scala/net/psforever/services/base/GenericEventBus.scala
index 72c2c43e3..74e929a91 100644
--- a/src/main/scala/net/psforever/services/base/GenericEventBus.scala
+++ b/src/main/scala/net/psforever/services/base/GenericEventBus.scala
@@ -6,9 +6,11 @@ import akka.util.Subclassification
 
 trait GenericEventBusMsg {
   def channel: String
+  def inner: Any
 }
 
-class GenericEventBus[A <: GenericEventBusMsg] extends ActorEventBus with SubchannelClassification {
+class GenericEventBus[A <: GenericEventBusMsg]
+  extends ActorEventBus with SubchannelClassification {
   type Event = A
   type Classifier = String
 
diff --git a/src/main/scala/net/psforever/services/base/GenericEventService.scala b/src/main/scala/net/psforever/services/base/GenericEventService.scala
index 6a4c030c0..a9c69a4df 100644
--- a/src/main/scala/net/psforever/services/base/GenericEventService.scala
+++ b/src/main/scala/net/psforever/services/base/GenericEventService.scala
@@ -12,6 +12,7 @@ trait GenericResponseEnvelope
   extends GenericEventBusMsg {
   def filter: PlanetSideGUID
   def reply: EventResponse
+  def inner: EventResponse = reply
 }
 
 trait GenericMessageEnvelope {
diff --git a/src/main/scala/net/psforever/services/base/GenericGuidEventBus.scala b/src/main/scala/net/psforever/services/base/GenericGuidEventBus.scala
new file mode 100644
index 000000000..16ed9d188
--- /dev/null
+++ b/src/main/scala/net/psforever/services/base/GenericGuidEventBus.scala
@@ -0,0 +1,106 @@
+// Copyright (c) 2026 PSForever
+package net.psforever.services.base
+
+import net.psforever.types.PlanetSideGUID
+import scala.collection.concurrent.{Map => CMap}
+import scala.jdk.CollectionConverters._
+import java.util.concurrent.ConcurrentHashMap
+import scala.annotation.unused
+
+/*
+Adapted from the rating limiting code in https://github.com/Pinapse/giant with permission
+ */
+
+trait GenericGuidEventBusMsg extends GenericEventBusMsg {
+  def guid: PlanetSideGUID
+}
+
+class RateLimitScheduler[A <: GenericGuidEventBusMsg](eventBus: GenericGuidEventBus[A], interval: Long) extends Thread {
+  private var hasWork: Boolean = false
+  private var working: Boolean = false
+  private var timeOfLastFlush: Long = 0L
+  private val buffer: CMap[String, CMap[String, CMap[PlanetSideGUID, A]]] =
+    new ConcurrentHashMap[String, CMap[String, CMap[PlanetSideGUID, A]]]().asScala
+
+  override def run(): Unit = {
+    while (working) {
+      //originally, there was a Thread.sleep(interval) here, replaced by timeOfLastFlush logic
+      flushBuffer()
+    }
+  }
+
+  def push(event: A): Unit = {
+    val eventClassName = event.inner.getClass.getName
+    val cache =
+      buffer
+        .getOrElseUpdate(event.channel, new ConcurrentHashMap[String, CMap[PlanetSideGUID, A]]().asScala)
+        .getOrElseUpdate(eventClassName, new ConcurrentHashMap[PlanetSideGUID, A]().asScala)
+    cache.updateWith(event.guid) { _ => Some(event) }
+    hasWork = true
+  }
+
+  def flushBuffer(): Unit = {
+    val curr = System.currentTimeMillis()
+    if (hasWork && timeOfLastFlush + interval <= curr) {
+      flushBufferNow(curr)
+    }
+  }
+
+  def flushBufferNow(curr: Long = System.currentTimeMillis()): Unit = {
+    buffer.foreachEntry { (_, map) =>
+      map.foreachEntry { (_, map) =>
+        map.foreachEntry { (_, event) =>
+          eventBus.publishForce(event)
+        }
+        map.clear()
+      }
+    }
+    hasWork = false
+    timeOfLastFlush = curr
+  }
+
+  override def start(): Unit = {
+    working = true
+    timeOfLastFlush = System.currentTimeMillis()
+    super.start()
+  }
+
+  def isRunning: Boolean = {
+    working
+  }
+
+  def stopRunning(): Unit = {
+    working = false
+    flushBufferNow()
+  }
+}
+
+abstract class GenericGuidEventBus[A <: GenericGuidEventBusMsg](rateLimit: Double)
+  extends GenericEventBus[A] {
+  private val rateLimitedDispatch = new RateLimitScheduler[A](
+    eventBus = this,
+    scala.math.floor(1000.0 / rateLimit).toInt
+  )
+  rateLimitedDispatch.start()
+
+  override def publish(event: Event): Unit = {
+    if (rateLimit > 0 && shouldRateLimit(event) && rateLimitedDispatch.isRunning) {
+      rateLimitedDispatch.push(event)
+    } else {
+      publishForce(event)
+    }
+  }
+
+  def shouldRateLimit(@unused event: Event): Boolean
+
+  def publishForce(event: Event): Unit = {
+    super.publish(event)
+  }
+
+//  override protected def publish(event: Event, subscriber: Subscriber): Unit = {
+//    val trimmedEventClassName =
+//      event.inner.getClass().getName().stripPrefix(event.inner.getClass.getPackageName() + ".").stripSuffix("$")
+//    GenericGuidEventBus.genericGuidEventBusPublish.labels(event.channel, trimmedEventClassName).inc()
+//    subscriber ! event
+//  }
+}