@Test public void publishPayloadCancelsCountDownLatchIfOneTaskFails() throws InterruptedException { final Executor realExecutor = Executors.newCachedThreadPool(); when(executorFactory.createCachedThreadPool()).thenReturn(realExecutor); asyncPublisher = new AsyncBatchPayloadPublisher(executorFactory, countDownLatchFactory, publisher, encoder); final EncodedPayload payload = mock(EncodedPayload.class); final EncodedPayload strippedPayload = 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); when(encoder.forRecipient(any(EncodedPayload.class), any(PublicKey.class))).thenReturn(strippedPayload); final PublishPayloadException cause = new PublishPayloadException("some exception"); doThrow(cause) .doNothing() .when(publisher).publishPayload(any(EncodedPayload.class), any(PublicKey.class)); doAnswer( invocation -> { Thread.sleep(200); return null; }) .when(countDownLatch) .await(); asyncPublisher.publishPayload(payload, recipients); verify(executorFactory, times(2)).createCachedThreadPool(); verify(countDownLatchFactory).create(2); verify(encoder).forRecipient(payload, recipient); verify(encoder).forRecipient(payload, otherRecipient); verify(publisher).publishPayload(strippedPayload, recipient); verify(publisher).publishPayload(strippedPayload, otherRecipient); verify(countDownLatch).countDown(); verify(countDownLatch).cancelWithException(cause); 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);  }