#2400 Use dedicated stream topic for stream dispatcher subscription

TheHive-Project · Jul 1, 2022 · 0e330c7 · 0e330c7
1 parent d281061
commit 0e330c7
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 6 deletions.
diff --git a/thehive/app/org/thp/thehive/services/AuditSrv.scala b/thehive/app/org/thp/thehive/services/AuditSrv.scala
@@ -105,7 +105,7 @@ class AuditSrv @Inject() (
           case Status.COMMIT =>
             logger.debug("Sending audit to stream bus and to notification actor")
             val auditIds = ids.map(_._2)
-            eventSrv.publish(StreamTopic())(AuditStreamMessage(auditIds: _*))
+            eventSrv.publish(StreamTopic.dispatcher)(AuditStreamMessage(auditIds: _*))
             notificationActor ! AuditNotificationMessage(auditIds: _*)
           case _ =>
         }

diff --git a/thehive/app/org/thp/thehive/services/FlowActor.scala b/thehive/app/org/thp/thehive/services/FlowActor.scala
@@ -50,8 +50,8 @@ class FlowActor extends Actor {
   def fromDate: Date = new Date(System.currentTimeMillis() - maxAgeConfig.get.toMillis)
 
   lazy val eventSrv: EventSrv   = injector.getInstance(classOf[EventSrv])
-  override def preStart(): Unit = eventSrv.subscribe(StreamTopic(), self)
-  override def postStop(): Unit = eventSrv.unsubscribe(StreamTopic(), self)
+  override def preStart(): Unit = eventSrv.subscribe(StreamTopic.dispatcher, self)
+  override def postStop(): Unit = eventSrv.unsubscribe(StreamTopic.dispatcher, self)
 
   def flowQuery(
       caseId: Option[EntityIdOrName]

diff --git a/thehive/app/org/thp/thehive/services/StreamSrv.scala b/thehive/app/org/thp/thehive/services/StreamSrv.scala
@@ -1,6 +1,6 @@
 package org.thp.thehive.services
 
-import akka.actor.{actorRef2Scala, Actor, ActorIdentity, ActorRef, ActorSystem, Cancellable, Identify, PoisonPill, Props}
+import akka.actor.{Actor, ActorIdentity, ActorRef, ActorSystem, Cancellable, Identify, PoisonPill, Props}
 import akka.pattern.{ask, AskTimeoutException}
 import akka.serialization.Serializer
 import akka.util.Timeout
@@ -26,7 +26,8 @@ import scala.util.{Random, Try}
 sealed trait StreamMessage extends Serializable
 
 object StreamTopic {
-  def apply(streamId: String = ""): String = if (streamId.isEmpty) "stream" else s"stream-$streamId"
+  def apply(streamId: String): String = s"stream-$streamId"
+  val dispatcher: String              = "stream"
 }
 
 case class AuditStreamMessage(id: EntityId*) extends StreamMessage
@@ -192,7 +193,7 @@ class StreamSrv @Inject() (
       )
     logger.debug(s"Register stream actor ${streamActor.path}")
     eventSrv.subscribe(StreamTopic(streamId), streamActor)
-    eventSrv.subscribe(StreamTopic(), streamActor)
+    eventSrv.subscribe(StreamTopic.dispatcher, streamActor)
     streamId
   }