ConcurrentCompositeLineConsumer implements LineConsumer { @Override public void writeLine(String line) { if (isOpen && lock.readLock().tryLock()) { try { for (LineConsumer lineConsumer : lineConsumers) { try { lineConsumer.writeLine(line); } catch (ConsumerAlreadyClosedException | ClosedByInterruptException e) { lineConsumers.remove( lineConsumer); if (lineConsumers.size() == 0) { isOpen = false; } } catch (IOException e) { LOG.error( String.format( "An error occurred while writing line to the line consumer %s", lineConsumer), e); } } } finally { lock.readLock().unlock(); } } } ConcurrentCompositeLineConsumer(LineConsumer... lineConsumers); boolean isOpen(); @Override void close(); @Override void writeLine(String line);  }
@Test public void shouldWriteMessageIntoEachConsumer() throws Exception { final String message = "Test line"; concurrentCompositeLineConsumer.writeLine(message); for (LineConsumer subConsumer : subConsumers) { verify(subConsumer).writeLine(eq(message)); } }
@Test(dataProvider = "subConsumersExceptions") public void shouldCloseSubConsumerOnException(Throwable exception) throws Exception { final String message = "Test line"; final String message2 = "Test line2"; LineConsumer closedConsumer = mock(LineConsumer.class); doThrow(exception).when(closedConsumer).writeLine(anyString()); concurrentCompositeLineConsumer = new ConcurrentCompositeLineConsumer(appendTo(subConsumers, closedConsumer)); concurrentCompositeLineConsumer.writeLine(message); concurrentCompositeLineConsumer.writeLine(message2); verify(closedConsumer, never()).writeLine(eq(message2)); for (LineConsumer consumer : subConsumers) { verify(consumer).writeLine(eq(message2)); } }
@Test public void shouldDoNothingOnWriteLineIfAllSubConsumersAreClosed() throws Exception { final String message = "Test line"; LineConsumer[] closedConsumers = subConsumers; for (LineConsumer consumer : closedConsumers) { doThrow(ConsumerAlreadyClosedException.class).when(consumer).writeLine(anyString()); } concurrentCompositeLineConsumer = new ConcurrentCompositeLineConsumer(closedConsumers); concurrentCompositeLineConsumer.writeLine("Error"); concurrentCompositeLineConsumer.writeLine(message); for (LineConsumer subConsumer : closedConsumers) { verify(subConsumer, never()).writeLine(eq(message)); } }
@Test public void shouldBeAbleToWriteIntoSubConsumersSimultaneously() throws Exception { final String message1 = "Message 1"; final String message2 = "Message 2"; WaitingAnswer<Void> waitingAnswer = new WaitingAnswer<>(); doAnswer(waitingAnswer).when(lineConsumer2).writeLine(eq(message1)); executor.execute(() -> concurrentCompositeLineConsumer.writeLine(message1)); waitingAnswer.waitAnswerCall(1, TimeUnit.SECONDS); concurrentCompositeLineConsumer.writeLine(message2); waitingAnswer.completeAnswer(); awaitFinalization(); for (LineConsumer consumer : subConsumers) { verify(consumer).writeLine(eq(message1)); verify(consumer).writeLine(eq(message2)); } }