@Test public void publishPayloadUsesThreadForEachRecipient() throws InterruptedException { final EncodedPayload payload = mock(EncodedPayload.class); final PublicKey recipient = PublicKey.from("RECIPIENT".getBytes()); final PublicKey otherRecipient = PublicKey.from("OTHERRECIPIENT".getBytes()); final List<PublicKey> recipients = List.of(recipient, otherRecipient); asyncPublisher.publishPayload(payload, recipients); verify(countDownLatchFactory).create(2); verify(executorFactory).createCachedThreadPool(); verify(executor, times(2)).execute(any(Runnable.class)); verify(countDownLatch).await(); }
@Override public void publishPayload(EncodedPayload payload, List<PublicKey> recipientKeys) { if (recipientKeys.size() == 0) { return; } final CancellableCountDownLatch latch = countDownLatchFactory.create(recipientKeys.size()); recipientKeys.forEach( recipient -> executor.execute( () -> { try { final EncodedPayload outgoing = encoder.forRecipient(payload, recipient); publisher.publishPayload(outgoing, recipient); latch.countDown(); } catch (RuntimeException e) { LOGGER.info("unable to publish payload in batch: {}", e.getMessage()); latch.cancelWithException(e); } })); try { latch.await(); } catch (InterruptedException e) { throw new BatchPublishPayloadException(e); } }
AsyncBatchPayloadPublisher implements BatchPayloadPublisher { @Override public void publishPayload(EncodedPayload payload, List<PublicKey> recipientKeys) { if (recipientKeys.size() == 0) { return; } final CancellableCountDownLatch latch = countDownLatchFactory.create(recipientKeys.size()); recipientKeys.forEach( recipient -> executor.execute( () -> { try { final EncodedPayload outgoing = encoder.forRecipient(payload, recipient); publisher.publishPayload(outgoing, recipient); latch.countDown(); } catch (RuntimeException e) { LOGGER.info("unable to publish payload in batch: {}", e.getMessage()); latch.cancelWithException(e); } })); try { latch.await(); } catch (InterruptedException e) { throw new BatchPublishPayloadException(e); } } }
AsyncBatchPayloadPublisher implements BatchPayloadPublisher { @Override public void publishPayload(EncodedPayload payload, List<PublicKey> recipientKeys) { if (recipientKeys.size() == 0) { return; } final CancellableCountDownLatch latch = countDownLatchFactory.create(recipientKeys.size()); recipientKeys.forEach( recipient -> executor.execute( () -> { try { final EncodedPayload outgoing = encoder.forRecipient(payload, recipient); publisher.publishPayload(outgoing, recipient); latch.countDown(); } catch (RuntimeException e) { LOGGER.info("unable to publish payload in batch: {}", e.getMessage()); latch.cancelWithException(e); } })); try { latch.await(); } catch (InterruptedException e) { throw new BatchPublishPayloadException(e); } } AsyncBatchPayloadPublisher(
            ExecutorFactory executorFactory,
            CancellableCountDownLatchFactory countDownLatchFactory,
            PayloadPublisher publisher,
            PayloadEncoder encoder); }
AsyncBatchPayloadPublisher implements BatchPayloadPublisher { @Override public void publishPayload(EncodedPayload payload, List<PublicKey> recipientKeys) { if (recipientKeys.size() == 0) { return; } final CancellableCountDownLatch latch = countDownLatchFactory.create(recipientKeys.size()); recipientKeys.forEach( recipient -> executor.execute( () -> { try { final EncodedPayload outgoing = encoder.forRecipient(payload, recipient); publisher.publishPayload(outgoing, recipient); latch.countDown(); } catch (RuntimeException e) { LOGGER.info("unable to publish payload in batch: {}", e.getMessage()); latch.cancelWithException(e); } })); try { latch.await(); } catch (InterruptedException e) { throw new BatchPublishPayloadException(e); } } AsyncBatchPayloadPublisher(
            ExecutorFactory executorFactory,
            CancellableCountDownLatchFactory countDownLatchFactory,
            PayloadPublisher publisher,
            PayloadEncoder encoder); @Override void publishPayload(EncodedPayload payload, List<PublicKey> recipientKeys); }
AsyncBatchPayloadPublisher implements BatchPayloadPublisher { @Override public void publishPayload(EncodedPayload payload, List<PublicKey> recipientKeys) { if (recipientKeys.size() == 0) { return; } final CancellableCountDownLatch latch = countDownLatchFactory.create(recipientKeys.size()); recipientKeys.forEach( recipient -> executor.execute( () -> { try { final EncodedPayload outgoing = encoder.forRecipient(payload, recipient); publisher.publishPayload(outgoing, recipient); latch.countDown(); } catch (RuntimeException e) { LOGGER.info("unable to publish payload in batch: {}", e.getMessage()); latch.cancelWithException(e); } })); try { latch.await(); } catch (InterruptedException e) { throw new BatchPublishPayloadException(e); } } AsyncBatchPayloadPublisher(
            ExecutorFactory executorFactory,
            CancellableCountDownLatchFactory countDownLatchFactory,
            PayloadPublisher publisher,
            PayloadEncoder encoder); @Override void publishPayload(EncodedPayload payload, List<PublicKey> recipientKeys);  }