@ Synchronized ( "lock" ) public void disconnect ( final ReactiveSeq < T > stream ) { Option < Queue < T >> o = streamToQueue . get ( stream ) ; distributor . removeQueue ( streamToQueue . getOrElse ( stream , new Queue <> ( ) ) ) ; this . streamToQueue = streamToQueue . remove ( stream ) ; this . index -- ; }
Topic will maintain a queue for each Subscribing Stream If a Stream is finished with a Topic it is good practice to disconnect from the Topic so messages will no longer be stored for that Stream