added reset all subscriptions functionality

common/src/main/scala/tmt/common/Messages.scala

@@ -8,4 +8,5 @@ object Messages {
   case class UpdateDelay(serverName: String, value: FiniteDuration)
   case class Subscribe(connection: Connection)
   case class Unsubscribe(connection: Connection)
+  case class ResetConnections()
 }

frontend/app/actors/ConnectionStore.scala

@@ -5,7 +5,7 @@ import akka.cluster.Cluster
 import akka.cluster.ClusterEvent._
 import akka.cluster.pubsub.DistributedPubSub
 import akka.cluster.pubsub.DistributedPubSubMediator.{Publish, Subscribe}
-import akka.persistence.{SnapshotOffer, PersistentActor}
+import akka.persistence.{SnapshotSelectionCriteria, SnapshotOffer, PersistentActor}
 import tmt.common.Messages
 import tmt.shared.Topics
 import tmt.shared.models.ConnectionSet
@@ -34,9 +34,10 @@ class ConnectionStore extends PersistentActor {
   }
 
   def receiveCommand = {
-    case event: Messages.Subscribe      => addConnection(event)
-    case event: Messages.Unsubscribe    => removeConnection(event)
-    case x: MemberUp                    =>
+    case event: Messages.Subscribe        => addConnection(event)
+    case event: Messages.Unsubscribe      => removeConnection(event)
+    case event: Messages.ResetConnections => connectionSet = ConnectionSet.empty; deleteSnapshots(SnapshotSelectionCriteria())
+    case x: MemberUp                      =>
       println("******: ", x)
       scheduler.scheduleOnce(2.second, mediator, Publish(Topics.Connections, connectionSet))
     case ConnectionStore.GetConnections => sender() ! connectionSet

frontend/app/controllers/StreamController.scala

@@ -54,4 +54,9 @@ class StreamController @Inject()(
     clusterClientService.unsubscribe(Connection(serverName, topic))
     Accepted("ok")
   }
+
+  def resetConnections() = Action {
+    clusterClientService.resetConnections()
+    Accepted("ok")
+  }
 }

frontend/app/services/ClusterClientService.scala

@@ -40,4 +40,8 @@ class ClusterClientService @Inject()(roleIndexService: RoleIndexService)(implici
   }
 
   def allConnections = (connectionStore ? ConnectionStore.GetConnections).mapTo[ConnectionSet]
+
+  def resetConnections() = {
+    mediator ! Publish(Topics.Subscription, Messages.ResetConnections())
+  }
 }

frontend/conf/routes

@@ -6,6 +6,7 @@ GET         /connections                       controllers.StreamController.conn
 POST        /:server/throttle/:delay           controllers.StreamController.throttle(server, delay: Long)
 POST        /:server/subscribe/:topic          controllers.StreamController.subscribe(server, topic)
 POST        /:server/unsubscribe/:topic        controllers.StreamController.unsubscribe(server, topic)
+POST        /reset-connections                 controllers.StreamController.resetConnections
 
 # Map static resources from the /public folder to the /assets URL path
 GET         /assets/*file                      controllers.Assets.at(path="/public", file)