Introduces multiple broker fixes

1. Unsubscribe from broker on actor stop
2. Support multiple subscriptions from the same actor type (previously a TODO)
3. Fixes a bug where broker always put self as an actor id.

Signed-off-by: Egor Ivkov <e.o.ivkov@gmail.com>

iroha/src/block_sync.rs

@@ -103,6 +103,10 @@ impl<S: SumeragiTrait, W: WorldTrait> Actor for BlockSynchronizer<S, W> {
         self.broker.subscribe::<ContinueSync, _>(ctx);
         ctx.notify_every::<ReceiveUpdates>(self.gossip_period);
     }
+
+    fn broker(&self) -> Option<&broker::Broker> {
+        Some(&self.broker)
+    }
 }
 
 #[async_trait::async_trait]

iroha/src/kura.rs

@@ -116,6 +116,10 @@ impl<W: WorldTrait> Actor for Kura<W> {
             }
         }
     }
+
+    fn broker(&self) -> Option<&broker::Broker> {
+        Some(&self.broker)
+    }
 }
 
 #[async_trait::async_trait]

iroha/src/queue.rs

@@ -84,6 +84,10 @@ impl<W: WorldTrait> Actor for Queue<W> {
         self.broker
             .subscribe::<VersionedAcceptedTransaction, _>(ctx);
     }
+
+    fn broker(&self) -> Option<&broker::Broker> {
+        Some(&self.broker)
+    }
 }
 
 #[async_trait::async_trait]

iroha/src/sumeragi/mod.rs

@@ -213,6 +213,10 @@ impl<Q: QueueTrait, G: GenesisNetworkTrait, W: WorldTrait> Actor for Sumeragi<Q,
         self.broker.subscribe::<Init, _>(ctx);
         self.broker.subscribe::<CommitBlock, _>(ctx);
     }
+
+    fn broker(&self) -> Option<&broker::Broker> {
+        Some(&self.broker)
+    }
 }
 
 #[async_trait::async_trait]

iroha/test_network/tests/sumeragi_with_mock.rs

@@ -600,7 +600,7 @@ async fn all_peers_commit_block() {
 
     let mut channels = network
         .peers()
-        .map(|peer| peer.broker.subscribe_with_channel::<Stored>())
+        .map(|peer| peer.broker.subscribe_with_channel::<Stored>().0)
         .collect::<Vec<_>>();
 
     // Send tx to leader
@@ -626,7 +626,7 @@ async fn change_view_on_commit_timeout() {
 
     let mut channels = network
         .peers()
-        .map(|peer| peer.broker.subscribe_with_channel::<Stored>())
+        .map(|peer| peer.broker.subscribe_with_channel::<Stored>().0)
         .collect::<Vec<_>>();
 
     // send to leader
@@ -664,7 +664,7 @@ async fn change_view_on_tx_receipt_timeout() {
 
     let mut channels = network
         .peers()
-        .map(|peer| peer.broker.subscribe_with_channel::<Stored>())
+        .map(|peer| peer.broker.subscribe_with_channel::<Stored>().0)
         .collect::<Vec<_>>();
 
     // send to not leader
@@ -706,7 +706,7 @@ async fn change_view_on_block_creation_timeout() {
 
     let mut channels = network
         .peers()
-        .map(|peer| peer.broker.subscribe_with_channel::<Stored>())
+        .map(|peer| peer.broker.subscribe_with_channel::<Stored>().0)
         .collect::<Vec<_>>();
 
     // send to not leader
@@ -740,7 +740,7 @@ async fn not_enough_votes() {
 
     let mut channels = network
         .peers()
-        .map(|peer| peer.broker.subscribe_with_channel::<Stored>())
+        .map(|peer| peer.broker.subscribe_with_channel::<Stored>().0)
         .collect::<Vec<_>>();
 
     // send to not leader