PeerGroup implements TransactionBroadcaster { public ListenableFuture<List<Peer>> waitForPeersOfVersion(final int numPeers, final long protocolVersion) { List<Peer> foundPeers = findPeersOfAtLeastVersion(protocolVersion); if (foundPeers.size() >= numPeers) { return Futures.immediateFuture(foundPeers); } final SettableFuture<List<Peer>> future = SettableFuture.create(); addConnectedEventListener(new PeerConnectedEventListener() { @Override public void onPeerConnected(Peer peer, int peerCount) { final List<Peer> peers = findPeersOfAtLeastVersion(protocolVersion); if (peers.size() >= numPeers) { future.set(peers); removeConnectedEventListener(this); } } }); return future; } PeerGroup(NetworkParameters params);  PeerGroup(Context context);  PeerGroup(NetworkParameters params, @Nullable AbstractBlockChain chain);  PeerGroup(Context context, @Nullable AbstractBlockChain chain);  PeerGroup(NetworkParameters params, @Nullable AbstractBlockChain chain, ClientConnectionManager connectionManager); private  PeerGroup(Context context, @Nullable AbstractBlockChain chain, ClientConnectionManager connectionManager); void setPeerDiscoveryTimeoutMillis(long peerDiscoveryTimeoutMillis); void setMaxConnections(int maxConnections); void setDownloadTxDependencies(int depth); int getMaxConnections(); void setVersionMessage(VersionMessage ver); VersionMessage getVersionMessage(); void setUserAgent(String name, String version, @Nullable String comments); void setUserAgent(String name, String version); @Deprecated @SuppressWarnings("deprecation") void addEventListener(AbstractPeerEventListener listener, Executor executor); @Deprecated @SuppressWarnings("deprecation") void addEventListener(AbstractPeerEventListener listener); void addBlocksDownloadedEventListener(BlocksDownloadedEventListener listener); void addBlocksDownloadedEventListener(Executor executor, BlocksDownloadedEventListener listener); void addChainDownloadStartedEventListener(ChainDownloadStartedEventListener listener); void addChainDownloadStartedEventListener(Executor executor, ChainDownloadStartedEventListener listener); void addConnectedEventListener(PeerConnectedEventListener listener); void addConnectedEventListener(Executor executor, PeerConnectedEventListener listener); void addDisconnectedEventListener(PeerDisconnectedEventListener listener); void addDisconnectedEventListener(Executor executor, PeerDisconnectedEventListener listener); void addDiscoveredEventListener(PeerDiscoveredEventListener listener); void addDiscoveredEventListener(Executor executor, PeerDiscoveredEventListener listener); void addGetDataEventListener(GetDataEventListener listener); void addGetDataEventListener(final Executor executor, final GetDataEventListener listener); void addOnTransactionBroadcastListener(OnTransactionBroadcastListener listener); void addOnTransactionBroadcastListener(Executor executor, OnTransactionBroadcastListener listener); void addPreMessageReceivedEventListener(PreMessageReceivedEventListener listener); void addPreMessageReceivedEventListener(Executor executor, PreMessageReceivedEventListener listener); @Deprecated @SuppressWarnings("deprecation") void removeEventListener(AbstractPeerEventListener listener); boolean removeBlocksDownloadedEventListener(BlocksDownloadedEventListener listener); boolean removeChainDownloadStartedEventListener(ChainDownloadStartedEventListener listener); boolean removeConnectedEventListener(PeerConnectedEventListener listener); boolean removeDisconnectedEventListener(PeerDisconnectedEventListener listener); boolean removeDiscoveredEventListener(PeerDiscoveredEventListener listener); boolean removeGetDataEventListener(GetDataEventListener listener); boolean removeOnTransactionBroadcastListener(OnTransactionBroadcastListener listener); boolean removePreMessageReceivedEventListener(PreMessageReceivedEventListener listener); List<Peer> getConnectedPeers(); List<Peer> getPendingPeers(); void addAddress(PeerAddress peerAddress); void setRequiredServices(long requiredServices); void addAddress(InetAddress address); void addPeerDiscovery(PeerDiscovery peerDiscovery); ListenableFuture startAsync(); void start(); ListenableFuture stopAsync(); void stop(); void addWallet(Wallet wallet); ListenableFuture<BloomFilter> addPeerFilterProvider(PeerFilterProvider provider); void removePeerFilterProvider(PeerFilterProvider provider); void removeWallet(Wallet wallet); ListenableFuture<BloomFilter> recalculateFastCatchupAndFilter(final FilterRecalculateMode mode); void setBloomFilterFalsePositiveRate(double bloomFilterFPRate); int numConnectedPeers(); @Nullable Peer connectTo(InetSocketAddress address); @Nullable Peer connectToLocalHost(); void setConnectTimeoutMillis(int connectTimeoutMillis); void startBlockChainDownload(PeerDataEventListener listener); void downloadBlockChain(); @Deprecated @Nullable TxConfidenceTable getMemoryPool(); void setFastCatchupTimeSecs(long secondsSinceEpoch); long getFastCatchupTimeSecs(); void setStallThreshold(int periodSecs, int bytesPerSecond); ListenableFuture<List<Peer>> waitForPeers(final int numPeers); ListenableFuture<List<Peer>> waitForPeersOfVersion(final int numPeers, final long protocolVersion); List<Peer> findPeersOfAtLeastVersion(long protocolVersion); ListenableFuture<List<Peer>> waitForPeersWithServiceMask(final int numPeers, final int mask); List<Peer> findPeersWithServiceMask(int mask); int getMinBroadcastConnections(); void setMinBroadcastConnections(int value); @Override TransactionBroadcast broadcastTransaction(final Transaction tx); TransactionBroadcast broadcastTransaction(final Transaction tx, final int minConnections); long getPingIntervalMsec(); void setPingIntervalMsec(long pingIntervalMsec); void setMinRequiredProtocolVersion(int minRequiredProtocolVersion); int getMinRequiredProtocolVersion(); int getMostCommonChainHeight(); static int getMostCommonChainHeight(final List<Peer> peers); Peer getDownloadPeer(); int getMaxPeersToDiscoverCount(); void setMaxPeersToDiscoverCount(int maxPeersToDiscoverCount); boolean getUseLocalhostPeerWhenPossible(); void setUseLocalhostPeerWhenPossible(boolean useLocalhostPeerWhenPossible); boolean isRunning(); void setBloomFilteringEnabled(boolean bloomFilteringEnabled); boolean isBloomFilteringEnabled(); static final int DEFAULT_CONNECTIONS; static final long DEFAULT_PING_INTERVAL_MSEC; static final double DEFAULT_BLOOM_FILTER_FP_RATE; static final double MAX_FP_RATE_INCREASE; static final int DEFAULT_CONNECT_TIMEOUT_MILLIS; }
@Test public void waitForPeersOfVersion() throws Exception { final int baseVer = peerGroup.getMinRequiredProtocolVersion() + 3000; final int newVer = baseVer + 1000; ListenableFuture<List<Peer>> future = peerGroup.waitForPeersOfVersion(2, newVer); VersionMessage ver1 = new VersionMessage(UNITTEST, 10); ver1.clientVersion = baseVer; ver1.localServices = VersionMessage.NODE_NETWORK; VersionMessage ver2 = new VersionMessage(UNITTEST, 10); ver2.clientVersion = newVer; ver2.localServices = VersionMessage.NODE_NETWORK; peerGroup.start(); assertFalse(future.isDone()); connectPeer(1, ver1); assertFalse(future.isDone()); connectPeer(2, ver2); assertFalse(future.isDone()); assertTrue(peerGroup.waitForPeersOfVersion(1, newVer).isDone()); connectPeer(3, ver2); future.get(); assertTrue(future.isDone()); }