mirror of https://github.com/hyperledger/besu
[PAN-2327] Notify of dropped messages (#1156)
* Added ability to subscribe to dropped transactions from the transaction pending pool. Implemented subscription webservice to support this. * Added metrics to the pending transactions, tracking the number of local and remote transactions in the pool. * Converted listener management in pending transactions to use the Subscribers util object. Signed-off-by: Adrian Sutton <adrian.sutton@consensys.net>pull/2/head
Rob Dawson
6 years ago
committed by
GitHub
parent
324222752a
commit
cdefb330be
@ -0,0 +1,19 @@ |
||||
/* |
||||
* Copyright 2018 ConsenSys AG. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on |
||||
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the |
||||
* specific language governing permissions and limitations under the License. |
||||
*/ |
||||
package tech.pegasys.pantheon.ethereum.core; |
||||
|
||||
@FunctionalInterface |
||||
public interface PendingTransactionDroppedListener { |
||||
|
||||
void onTransactionDropped(Transaction transaction); |
||||
} |
||||
@ -0,0 +1,52 @@ |
||||
/* |
||||
* Copyright 2018 ConsenSys AG. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on |
||||
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the |
||||
* specific language governing permissions and limitations under the License. |
||||
*/ |
||||
package tech.pegasys.pantheon.ethereum.jsonrpc.websocket.subscription.pending; |
||||
|
||||
import tech.pegasys.pantheon.ethereum.core.Hash; |
||||
import tech.pegasys.pantheon.ethereum.core.PendingTransactionDroppedListener; |
||||
import tech.pegasys.pantheon.ethereum.core.Transaction; |
||||
import tech.pegasys.pantheon.ethereum.jsonrpc.websocket.subscription.Subscription; |
||||
import tech.pegasys.pantheon.ethereum.jsonrpc.websocket.subscription.SubscriptionManager; |
||||
import tech.pegasys.pantheon.ethereum.jsonrpc.websocket.subscription.request.SubscriptionType; |
||||
|
||||
import java.util.List; |
||||
|
||||
public class PendingTransactionDroppedSubscriptionService |
||||
implements PendingTransactionDroppedListener { |
||||
|
||||
private final SubscriptionManager subscriptionManager; |
||||
|
||||
public PendingTransactionDroppedSubscriptionService( |
||||
final SubscriptionManager subscriptionManager) { |
||||
this.subscriptionManager = subscriptionManager; |
||||
} |
||||
|
||||
@Override |
||||
public void onTransactionDropped(final Transaction transaction) { |
||||
notifySubscribers(transaction.hash()); |
||||
} |
||||
|
||||
private void notifySubscribers(final Hash pendingTransaction) { |
||||
final List<Subscription> subscriptions = pendingDroppedTransactionSubscriptions(); |
||||
|
||||
final PendingTransactionResult msg = new PendingTransactionResult(pendingTransaction); |
||||
for (final Subscription subscription : subscriptions) { |
||||
subscriptionManager.sendMessage(subscription.getId(), msg); |
||||
} |
||||
} |
||||
|
||||
private List<Subscription> pendingDroppedTransactionSubscriptions() { |
||||
return subscriptionManager.subscriptionsOfType( |
||||
SubscriptionType.DROPPED_PENDING_TRANSACTIONS, Subscription.class); |
||||
} |
||||
} |
||||
@ -0,0 +1,109 @@ |
||||
/* |
||||
* Copyright 2018 ConsenSys AG. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on |
||||
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the |
||||
* specific language governing permissions and limitations under the License. |
||||
*/ |
||||
package tech.pegasys.pantheon.ethereum.jsonrpc.websocket.subscription.pending; |
||||
|
||||
import static org.mockito.ArgumentMatchers.any; |
||||
import static org.mockito.ArgumentMatchers.eq; |
||||
import static org.mockito.ArgumentMatchers.refEq; |
||||
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 static org.mockito.Mockito.when; |
||||
|
||||
import tech.pegasys.pantheon.ethereum.chain.Blockchain; |
||||
import tech.pegasys.pantheon.ethereum.core.Block; |
||||
import tech.pegasys.pantheon.ethereum.core.Hash; |
||||
import tech.pegasys.pantheon.ethereum.core.Transaction; |
||||
import tech.pegasys.pantheon.ethereum.jsonrpc.websocket.subscription.Subscription; |
||||
import tech.pegasys.pantheon.ethereum.jsonrpc.websocket.subscription.SubscriptionManager; |
||||
import tech.pegasys.pantheon.ethereum.jsonrpc.websocket.subscription.request.SubscriptionType; |
||||
|
||||
import java.util.Arrays; |
||||
import java.util.HashMap; |
||||
import java.util.Map; |
||||
import java.util.stream.Collectors; |
||||
|
||||
import org.junit.Before; |
||||
import org.junit.Test; |
||||
import org.junit.runner.RunWith; |
||||
import org.mockito.Mock; |
||||
import org.mockito.junit.MockitoJUnitRunner; |
||||
|
||||
@RunWith(MockitoJUnitRunner.class) |
||||
public class PendingTransactionDroppedSubscriptionServiceTest { |
||||
|
||||
private static final Hash TX_ONE = |
||||
Hash.fromHexString("0x15876958423545c3c7b0fcf9be8ffb543305ee1b43db87ed380dcf0cd16589f7"); |
||||
|
||||
@Mock private SubscriptionManager subscriptionManager; |
||||
@Mock private Blockchain blockchain; |
||||
@Mock private Block block; |
||||
|
||||
private PendingTransactionDroppedSubscriptionService service; |
||||
|
||||
@Before |
||||
public void setUp() { |
||||
service = new PendingTransactionDroppedSubscriptionService(subscriptionManager); |
||||
} |
||||
|
||||
@Test |
||||
public void onTransactionAddedMustSendMessage() { |
||||
final long[] subscriptionIds = new long[] {5, 56, 989}; |
||||
setUpSubscriptions(subscriptionIds); |
||||
final Transaction pending = transaction(TX_ONE); |
||||
|
||||
service.onTransactionDropped(pending); |
||||
|
||||
verifyZeroInteractions(block); |
||||
verifyZeroInteractions(blockchain); |
||||
verifySubscriptionMangerInteractions(messages(TX_ONE, subscriptionIds)); |
||||
} |
||||
|
||||
private void verifySubscriptionMangerInteractions(final Map<Long, Hash> expected) { |
||||
verify(subscriptionManager) |
||||
.subscriptionsOfType(SubscriptionType.DROPPED_PENDING_TRANSACTIONS, Subscription.class); |
||||
|
||||
for (final Map.Entry<Long, Hash> message : expected.entrySet()) { |
||||
verify(subscriptionManager) |
||||
.sendMessage( |
||||
eq(message.getKey()), refEq(new PendingTransactionResult(message.getValue()))); |
||||
} |
||||
|
||||
verifyNoMoreInteractions(subscriptionManager); |
||||
} |
||||
|
||||
private Map<Long, Hash> messages(final Hash result, final long... subscriptionIds) { |
||||
final Map<Long, Hash> messages = new HashMap<>(); |
||||
|
||||
for (final long subscriptionId : subscriptionIds) { |
||||
messages.put(subscriptionId, result); |
||||
} |
||||
|
||||
return messages; |
||||
} |
||||
|
||||
private Transaction transaction(final Hash hash) { |
||||
final Transaction tx = mock(Transaction.class); |
||||
when(tx.hash()).thenReturn(hash); |
||||
return tx; |
||||
} |
||||
|
||||
private void setUpSubscriptions(final long... subscriptionsIds) { |
||||
when(subscriptionManager.subscriptionsOfType(any(), any())) |
||||
.thenReturn( |
||||
Arrays.stream(subscriptionsIds) |
||||
.mapToObj(id -> new Subscription(id, SubscriptionType.DROPPED_PENDING_TRANSACTIONS)) |
||||
.collect(Collectors.toList())); |
||||
} |
||||
} |
||||
Loading…
Reference in new issue