CompositeLineConsumer implements LineConsumer { @Override public void writeLine(String line) { if (isOpen) { for (LineConsumer lineConsumer : lineConsumers) { try { lineConsumer.writeLine(line); } catch (ClosedByInterruptException interrupted) { Thread.currentThread().interrupt(); isOpen = false; return; } catch (ConsumerAlreadyClosedException 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); } } } } CompositeLineConsumer(LineConsumer... lineConsumers); @Override void close(); @Override void writeLine(String line);  }
@Test public void shouldWriteMessageIntoEachConsumer() throws Exception { final String message = "Test line"; compositeLineConsumer.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()); compositeLineConsumer = new CompositeLineConsumer(appendTo(subConsumers, closedConsumer)); compositeLineConsumer.writeLine(message); compositeLineConsumer.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()); } compositeLineConsumer = new CompositeLineConsumer(closedConsumers); compositeLineConsumer.writeLine("Error"); compositeLineConsumer.writeLine(message); for (LineConsumer subConsumer : closedConsumers) { verify(subConsumer, never()).writeLine(eq(message)); } }
@Test public void stopsWritingOnceInterrupted() throws Exception { doThrow(new ClosedByInterruptException()).when(lineConsumer2).writeLine("test"); compositeLineConsumer.writeLine("test"); assertTrue(Thread.interrupted()); verify(lineConsumer1).writeLine("test"); verify(lineConsumer2).writeLine("test"); verify(lineConsumer3, never()).writeLine("test"); }