[PAN-3083] Transactions listeners should use the subscriber pattern

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/transactions/PendingTransactions.java

@@ -63,7 +63,8 @@ public class PendingTransactions {
   private final Map<Address, SortedMap<Long, TransactionInfo>> transactionsBySender =
       new HashMap<>();
 
-  private final Subscribers<PendingTransactionListener> listeners = Subscribers.create();
+  private final Subscribers<PendingTransactionListener> pendingTransactionSubscribers =
+      Subscribers.create();
 
   private final Subscribers<PendingTransactionDroppedListener> transactionDroppedListeners =
       Subscribers.create();
@@ -261,7 +262,7 @@ private boolean shouldReplace(
   }
 
   private void notifyTransactionAdded(final Transaction transaction) {
-    listeners.forEach(listener -> listener.onTransactionAdded(transaction));
+    pendingTransactionSubscribers.forEach(listener -> listener.onTransactionAdded(transaction));
   }
 
   private void notifyTransactionDropped(final Transaction transaction) {
@@ -289,12 +290,20 @@ public Set<TransactionInfo> getTransactionInfo() {
     return new HashSet<>(pendingTransactions.values());
   }
 
-  void addTransactionListener(final PendingTransactionListener listener) {
-    listeners.subscribe(listener);
+  long subscribePendingTransactions(final PendingTransactionListener listener) {
+    return pendingTransactionSubscribers.subscribe(listener);
+  }
+
+  void unsubscribePendingTransactions(final long id) {
+    pendingTransactionSubscribers.unsubscribe(id);
+  }
+
+  long subscribeDroppedTransactions(final PendingTransactionDroppedListener listener) {
+    return transactionDroppedListeners.subscribe(listener);
   }
 
-  void addTransactionDroppedListener(final PendingTransactionDroppedListener listener) {
-    transactionDroppedListeners.subscribe(listener);
+  void unsubscribeDroppedTransactions(final long id) {
+    transactionDroppedListeners.unsubscribe(id);
   }
 
   public OptionalLong getNextNonceForSender(final Address sender) {

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/transactions/TransactionPool.java

@@ -162,12 +162,20 @@ public void addRemoteTransactions(final Collection<Transaction> transactions) {
     }
   }
 
-  public void addTransactionListener(final PendingTransactionListener listener) {
-    pendingTransactions.addTransactionListener(listener);
+  public long subscribePendingTransactions(final PendingTransactionListener listener) {
+    return pendingTransactions.subscribePendingTransactions(listener);
   }
 
-  public void addTransactionDroppedListener(final PendingTransactionDroppedListener listener) {
-    pendingTransactions.addTransactionDroppedListener(listener);
+  public void unsubscribePendingTransactions(final long id) {
+    pendingTransactions.unsubscribePendingTransactions(id);
+  }
+
+  public long subscribeDroppedTransactions(final PendingTransactionDroppedListener listener) {
+    return pendingTransactions.subscribeDroppedTransactions(listener);
+  }
+
+  public void unsubscribeDroppedTransactions(final long id) {
+    pendingTransactions.unsubscribeDroppedTransactions(id);
   }
 
   @Override

ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/transactions/PendingTransactionsTest.java

@@ -15,6 +15,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.verifyZeroInteractions;
 
 import tech.pegasys.pantheon.crypto.SECP256K1.KeyPair;
@@ -159,16 +160,31 @@ public void shouldStartDroppingLocalTransactionsWhenPoolIsFullOfLocalTransaction
 
   @Test
   public void shouldNotifyListenerWhenRemoteTransactionAdded() {
-    transactions.addTransactionListener(listener);
+    transactions.subscribePendingTransactions(listener);
 
     transactions.addRemoteTransaction(transaction1);
 
     verify(listener).onTransactionAdded(transaction1);
   }
 
+  @Test
+  public void shouldNotNotifyListenerAfterUnsubscribe() {
+    final long id = transactions.subscribePendingTransactions(listener);
+
+    transactions.addRemoteTransaction(transaction1);
+
+    verify(listener).onTransactionAdded(transaction1);
+
+    transactions.unsubscribePendingTransactions(id);
+
+    transactions.addRemoteTransaction(transaction2);
+
+    verifyZeroInteractions(listener);
+  }
+
   @Test
   public void shouldNotifyListenerWhenLocalTransactionAdded() {
-    transactions.addTransactionListener(listener);
+    transactions.subscribePendingTransactions(listener);
 
     transactions.addLocalTransaction(transaction1);
 
@@ -179,18 +195,36 @@ public void shouldNotifyListenerWhenLocalTransactionAdded() {
   public void shouldNotifyDroppedListenerWhenRemoteTransactionDropped() {
     transactions.addRemoteTransaction(transaction1);
 
-    transactions.addTransactionDroppedListener(droppedListener);
+    transactions.subscribeDroppedTransactions(droppedListener);
 
     transactions.removeTransaction(transaction1);
 
     verify(droppedListener).onTransactionDropped(transaction1);
   }
 
+  @Test
+  public void shouldNotNotifyDroppedListenerAfterUnsubscribe() {
+    transactions.addRemoteTransaction(transaction1);
+    transactions.addRemoteTransaction(transaction2);
+
+    final long id = transactions.subscribeDroppedTransactions(droppedListener);
+
+    transactions.removeTransaction(transaction1);
+
+    verify(droppedListener).onTransactionDropped(transaction1);
+
+    transactions.unsubscribeDroppedTransactions(id);
+
+    transactions.removeTransaction(transaction2);
+
+    verifyNoMoreInteractions(droppedListener);
+  }
+
   @Test
   public void shouldNotifyDroppedListenerWhenLocalTransactionDropped() {
     transactions.addLocalTransaction(transaction1);
 
-    transactions.addTransactionDroppedListener(droppedListener);
+    transactions.subscribeDroppedTransactions(droppedListener);
 
     transactions.removeTransaction(transaction1);
 
@@ -201,7 +235,7 @@ public void shouldNotifyDroppedListenerWhenLocalTransactionDropped() {
   public void shouldNotNotifyDroppedListenerWhenTransactionAddedToBlock() {
     transactions.addRemoteTransaction(transaction1);
 
-    transactions.addTransactionDroppedListener(droppedListener);
+    transactions.subscribeDroppedTransactions(droppedListener);
 
     transactions.transactionAddedToBlock(transaction1);
 
@@ -310,7 +344,7 @@ public void shouldNotReplaceTransactionWithSameSenderAndNonceWhenGasPriceIsLower
     final Transaction transaction1b = transactionWithNonceSenderAndGasPrice(1, KEYS1, 1);
     assertThat(transactions.addRemoteTransaction(transaction1)).isTrue();
 
-    transactions.addTransactionListener(listener);
+    transactions.subscribePendingTransactions(listener);
     assertThat(transactions.addRemoteTransaction(transaction1b)).isFalse();
 
     assertTransactionNotPending(transaction1b);

ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/transactions/TransactionPoolTest.java

@@ -224,7 +224,7 @@ public void shouldReadTransactionsFromThePreviousCanonicalHeadWhenAReorgOccurs()
     final Block reorgFork1 = appendBlock(UInt256.ONE, commonParent);
     verifyChainHeadIs(originalFork2);
 
-    transactions.addTransactionListener(listener);
+    transactions.subscribePendingTransactions(listener);
     final Block reorgFork2 = appendBlock(UInt256.of(2000), reorgFork1.getHeader());
     verifyChainHeadIs(reorgFork2);
 

ethereum/jsonrpc/src/main/java/tech/pegasys/pantheon/ethereum/jsonrpc/internal/filter/FilterManager.java

@@ -48,7 +48,7 @@ public FilterManager(
     this.filterRepository = filterRepository;
     checkNotNull(blockchainQueries.getBlockchain());
     blockchainQueries.getBlockchain().observeBlockAdded(this::recordBlockEvent);
-    transactionPool.addTransactionListener(this::recordPendingTransactionEvent);
+    transactionPool.subscribePendingTransactions(this::recordPendingTransactionEvent);
     this.blockchainQueries = blockchainQueries;
   }
 

pantheon/src/main/java/tech/pegasys/pantheon/RunnerBuilder.java

@@ -598,8 +598,8 @@ private SubscriptionManager createSubscriptionManager(
         new PendingTransactionSubscriptionService(subscriptionManager);
     final PendingTransactionDroppedSubscriptionService pendingTransactionsRemoved =
         new PendingTransactionDroppedSubscriptionService(subscriptionManager);
-    transactionPool.addTransactionListener(pendingTransactions);
-    transactionPool.addTransactionDroppedListener(pendingTransactionsRemoved);
+    transactionPool.subscribePendingTransactions(pendingTransactions);
+    transactionPool.subscribeDroppedTransactions(pendingTransactionsRemoved);
     vertx.deployVerticle(subscriptionManager);
 
     return subscriptionManager;