@Test public void jaxRsProcessingExceptionRemovesNode() { ProcessingException processingException = new ProcessingException("OUCH"); CompletionException completionException = new CompletionException(processingException); when(p2pClient.sendPartyInfo(anyString(), any(byte[].class))).thenThrow(completionException); String uriData = "http: when(partyStore.getParties()).thenReturn(Set.of(URI.create(uriData))); partyInfoBroadcaster.pollSingleParty(uriData, "somebytes".getBytes()); verify(discovery).onDisconnect(URI.create(uriData)); verify(partyStore).remove(URI.create(uriData)); verify(p2pClient).sendPartyInfo(anyString(), any(byte[].class)); }
protected void pollSingleParty(final String url, final byte[] encodedPartyInfo) { final NodeUri nodeUri = NodeUri.create(url); CompletableFuture.runAsync( () -> { LOGGER.debug("Sending party info to {}", nodeUri.asString()); p2pClient.sendPartyInfo(url, encodedPartyInfo); LOGGER.debug("Sent party info to {}", nodeUri.asString()); }, executor) .exceptionally( ex -> { Throwable cause = Optional.of(ex).map(Throwable::getCause).orElse(ex); LOGGER.warn("Failed to connect to node {}, due to {}", url, cause.getMessage()); LOGGER.debug("Send failure exception", cause); if (ProcessingException.class.isInstance(cause)) { discovery.onDisconnect(URI.create(url)); partyStore.remove(URI.create(url)); } return null; }); }
PartyInfoBroadcaster implements Runnable { protected void pollSingleParty(final String url, final byte[] encodedPartyInfo) { final NodeUri nodeUri = NodeUri.create(url); CompletableFuture.runAsync( () -> { LOGGER.debug("Sending party info to {}", nodeUri.asString()); p2pClient.sendPartyInfo(url, encodedPartyInfo); LOGGER.debug("Sent party info to {}", nodeUri.asString()); }, executor) .exceptionally( ex -> { Throwable cause = Optional.of(ex).map(Throwable::getCause).orElse(ex); LOGGER.warn("Failed to connect to node {}, due to {}", url, cause.getMessage()); LOGGER.debug("Send failure exception", cause); if (ProcessingException.class.isInstance(cause)) { discovery.onDisconnect(URI.create(url)); partyStore.remove(URI.create(url)); } return null; }); } }
PartyInfoBroadcaster implements Runnable { protected void pollSingleParty(final String url, final byte[] encodedPartyInfo) { final NodeUri nodeUri = NodeUri.create(url); CompletableFuture.runAsync( () -> { LOGGER.debug("Sending party info to {}", nodeUri.asString()); p2pClient.sendPartyInfo(url, encodedPartyInfo); LOGGER.debug("Sent party info to {}", nodeUri.asString()); }, executor) .exceptionally( ex -> { Throwable cause = Optional.of(ex).map(Throwable::getCause).orElse(ex); LOGGER.warn("Failed to connect to node {}, due to {}", url, cause.getMessage()); LOGGER.debug("Send failure exception", cause); if (ProcessingException.class.isInstance(cause)) { discovery.onDisconnect(URI.create(url)); partyStore.remove(URI.create(url)); } return null; }); } PartyInfoBroadcaster(final P2pClient p2pClient);  PartyInfoBroadcaster(
        final Discovery discovery,
        final PartyInfoParser partyInfoParser,
        final P2pClient p2pClient,
        final Executor executor,
        final PartyStore partyStore); }
PartyInfoBroadcaster implements Runnable { protected void pollSingleParty(final String url, final byte[] encodedPartyInfo) { final NodeUri nodeUri = NodeUri.create(url); CompletableFuture.runAsync( () -> { LOGGER.debug("Sending party info to {}", nodeUri.asString()); p2pClient.sendPartyInfo(url, encodedPartyInfo); LOGGER.debug("Sent party info to {}", nodeUri.asString()); }, executor) .exceptionally( ex -> { Throwable cause = Optional.of(ex).map(Throwable::getCause).orElse(ex); LOGGER.warn("Failed to connect to node {}, due to {}", url, cause.getMessage()); LOGGER.debug("Send failure exception", cause); if (ProcessingException.class.isInstance(cause)) { discovery.onDisconnect(URI.create(url)); partyStore.remove(URI.create(url)); } return null; }); } PartyInfoBroadcaster(final P2pClient p2pClient);  PartyInfoBroadcaster(
        final Discovery discovery,
        final PartyInfoParser partyInfoParser,
        final P2pClient p2pClient,
        final Executor executor,
        final PartyStore partyStore); @Override void run(); }
PartyInfoBroadcaster implements Runnable { protected void pollSingleParty(final String url, final byte[] encodedPartyInfo) { final NodeUri nodeUri = NodeUri.create(url); CompletableFuture.runAsync( () -> { LOGGER.debug("Sending party info to {}", nodeUri.asString()); p2pClient.sendPartyInfo(url, encodedPartyInfo); LOGGER.debug("Sent party info to {}", nodeUri.asString()); }, executor) .exceptionally( ex -> { Throwable cause = Optional.of(ex).map(Throwable::getCause).orElse(ex); LOGGER.warn("Failed to connect to node {}, due to {}", url, cause.getMessage()); LOGGER.debug("Send failure exception", cause); if (ProcessingException.class.isInstance(cause)) { discovery.onDisconnect(URI.create(url)); partyStore.remove(URI.create(url)); } return null; }); } PartyInfoBroadcaster(final P2pClient p2pClient);  PartyInfoBroadcaster(
        final Discovery discovery,
        final PartyInfoParser partyInfoParser,
        final P2pClient p2pClient,
        final Executor executor,
        final PartyStore partyStore); @Override void run();  }