@Test public void handleConnectionError() { final String targetUri = "http: final PublicKey recipientKey = mock(PublicKey.class); Recipient recipient = mock(Recipient.class); when(recipient.getKey()).thenReturn(recipientKey); when(recipient.getUrl()).thenReturn(targetUri); NodeInfo nodeInfo = mock(NodeInfo.class); when(nodeInfo.getRecipients()).thenReturn(Set.of(recipient)); when(discovery.getCurrent()).thenReturn(nodeInfo); Client client = mock(Client.class); when(client.target(targetUri)).thenThrow(ProcessingException.class); final EncodedPayload payload = mock(EncodedPayload.class); when(encoder.encode(payload)).thenReturn("SomeData".getBytes()); RestPayloadPublisher restPayloadPublisher = new RestPayloadPublisher(client,encoder,discovery); try { restPayloadPublisher.publishPayload(payload, recipientKey); failBecauseExceptionWasNotThrown(NodeOfflineException.class); } catch (NodeOfflineException ex) { assertThat(ex).hasMessageContaining(targetUri); verify(client).target(targetUri); verify(discovery).getCurrent(); verify(encoder).encode(payload); } }
@Override public void publishPayload(EncodedPayload payload, PublicKey recipientKey) { final Recipient retrievedRecipientFromStore = discovery.getCurrent().getRecipients().stream() .filter(recipient -> recipientKey.equals(recipient.getKey())) .findAny() .orElseThrow( () -> new KeyNotFoundException( "Recipient not found for key: " + recipientKey.encodeToBase64())); final String targetUrl = retrievedRecipientFromStore.getUrl(); LOGGER.info("Publishing message to {}", targetUrl); final byte[] encoded = payloadEncoder.encode(payload); try (Response response = restclient .target(targetUrl) .path("/push") .request() .post(Entity.entity(encoded, MediaType.APPLICATION_OCTET_STREAM_TYPE))) { if (Response.Status.OK.getStatusCode() != response.getStatus() && Response.Status.CREATED.getStatusCode() != response.getStatus()) { throw new PublishPayloadException("Unable to push payload to recipient url " + targetUrl); } LOGGER.info("Published to {}", targetUrl); } catch (ProcessingException ex) { LOGGER.debug("",ex); throw new NodeOfflineException(URI.create(targetUrl)); } }
RestPayloadPublisher implements PayloadPublisher { @Override public void publishPayload(EncodedPayload payload, PublicKey recipientKey) { final Recipient retrievedRecipientFromStore = discovery.getCurrent().getRecipients().stream() .filter(recipient -> recipientKey.equals(recipient.getKey())) .findAny() .orElseThrow( () -> new KeyNotFoundException( "Recipient not found for key: " + recipientKey.encodeToBase64())); final String targetUrl = retrievedRecipientFromStore.getUrl(); LOGGER.info("Publishing message to {}", targetUrl); final byte[] encoded = payloadEncoder.encode(payload); try (Response response = restclient .target(targetUrl) .path("/push") .request() .post(Entity.entity(encoded, MediaType.APPLICATION_OCTET_STREAM_TYPE))) { if (Response.Status.OK.getStatusCode() != response.getStatus() && Response.Status.CREATED.getStatusCode() != response.getStatus()) { throw new PublishPayloadException("Unable to push payload to recipient url " + targetUrl); } LOGGER.info("Published to {}", targetUrl); } catch (ProcessingException ex) { LOGGER.debug("",ex); throw new NodeOfflineException(URI.create(targetUrl)); } } }
RestPayloadPublisher implements PayloadPublisher { @Override public void publishPayload(EncodedPayload payload, PublicKey recipientKey) { final Recipient retrievedRecipientFromStore = discovery.getCurrent().getRecipients().stream() .filter(recipient -> recipientKey.equals(recipient.getKey())) .findAny() .orElseThrow( () -> new KeyNotFoundException( "Recipient not found for key: " + recipientKey.encodeToBase64())); final String targetUrl = retrievedRecipientFromStore.getUrl(); LOGGER.info("Publishing message to {}", targetUrl); final byte[] encoded = payloadEncoder.encode(payload); try (Response response = restclient .target(targetUrl) .path("/push") .request() .post(Entity.entity(encoded, MediaType.APPLICATION_OCTET_STREAM_TYPE))) { if (Response.Status.OK.getStatusCode() != response.getStatus() && Response.Status.CREATED.getStatusCode() != response.getStatus()) { throw new PublishPayloadException("Unable to push payload to recipient url " + targetUrl); } LOGGER.info("Published to {}", targetUrl); } catch (ProcessingException ex) { LOGGER.debug("",ex); throw new NodeOfflineException(URI.create(targetUrl)); } } RestPayloadPublisher(Client restclient, Discovery discovery);  RestPayloadPublisher(Client restclient, PayloadEncoder payloadEncoder, Discovery discovery); }
RestPayloadPublisher implements PayloadPublisher { @Override public void publishPayload(EncodedPayload payload, PublicKey recipientKey) { final Recipient retrievedRecipientFromStore = discovery.getCurrent().getRecipients().stream() .filter(recipient -> recipientKey.equals(recipient.getKey())) .findAny() .orElseThrow( () -> new KeyNotFoundException( "Recipient not found for key: " + recipientKey.encodeToBase64())); final String targetUrl = retrievedRecipientFromStore.getUrl(); LOGGER.info("Publishing message to {}", targetUrl); final byte[] encoded = payloadEncoder.encode(payload); try (Response response = restclient .target(targetUrl) .path("/push") .request() .post(Entity.entity(encoded, MediaType.APPLICATION_OCTET_STREAM_TYPE))) { if (Response.Status.OK.getStatusCode() != response.getStatus() && Response.Status.CREATED.getStatusCode() != response.getStatus()) { throw new PublishPayloadException("Unable to push payload to recipient url " + targetUrl); } LOGGER.info("Published to {}", targetUrl); } catch (ProcessingException ex) { LOGGER.debug("",ex); throw new NodeOfflineException(URI.create(targetUrl)); } } RestPayloadPublisher(Client restclient, Discovery discovery);  RestPayloadPublisher(Client restclient, PayloadEncoder payloadEncoder, Discovery discovery); @Override void publishPayload(EncodedPayload payload, PublicKey recipientKey); }
RestPayloadPublisher implements PayloadPublisher { @Override public void publishPayload(EncodedPayload payload, PublicKey recipientKey) { final Recipient retrievedRecipientFromStore = discovery.getCurrent().getRecipients().stream() .filter(recipient -> recipientKey.equals(recipient.getKey())) .findAny() .orElseThrow( () -> new KeyNotFoundException( "Recipient not found for key: " + recipientKey.encodeToBase64())); final String targetUrl = retrievedRecipientFromStore.getUrl(); LOGGER.info("Publishing message to {}", targetUrl); final byte[] encoded = payloadEncoder.encode(payload); try (Response response = restclient .target(targetUrl) .path("/push") .request() .post(Entity.entity(encoded, MediaType.APPLICATION_OCTET_STREAM_TYPE))) { if (Response.Status.OK.getStatusCode() != response.getStatus() && Response.Status.CREATED.getStatusCode() != response.getStatus()) { throw new PublishPayloadException("Unable to push payload to recipient url " + targetUrl); } LOGGER.info("Published to {}", targetUrl); } catch (ProcessingException ex) { LOGGER.debug("",ex); throw new NodeOfflineException(URI.create(targetUrl)); } } RestPayloadPublisher(Client restclient, Discovery discovery);  RestPayloadPublisher(Client restclient, PayloadEncoder payloadEncoder, Discovery discovery); @Override void publishPayload(EncodedPayload payload, PublicKey recipientKey);  }