### Question:
AkkaStreamsDemo implements ReactiveStreamsDemo { public void printErrors() { final ActorSystem system = ActorSystem.create("reactive-messages"); final Materializer mat = ActorMaterializer.create(system); Source<String, NotUsed> messages = getMessages(); final Source<String, NotUsed> errors = messages .filter(m -> m.startsWith("Error")) .buffer(200, OverflowStrategy.fail()) .alsoTo(Sink.foreach(messageList::add)).map(m -> m.toString()); errors.runWith(Sink.foreach(System.out::println), mat); }  static List<Integer> doSquares(); static List<Integer> doParallelSquares(); static CompletionStage<ArrayList<Integer>> parallelSquares(); static Materializer createMaterializer(); @Override java.util.concurrent.Future<List<Integer>> doSquaresAsync(int count); @Override java.util.concurrent.Future<String> doStringConcatAsync(int count); @Override java.util.concurrent.Future<List<Integer>> doParallelSquaresAsync(int count); @Override java.util.concurrent.Future<String> doParallelStringConcatAsync(int count); void printErrors(); Source<String, NotUsed> getMessages(); void setChannel(Channel channel); Sink<String, CompletionStage<IOResult>> lineSink(String filename); void saveTextFile(List<String> text); Graph<SinkShape<String>, NotUsed> createFileSinkGraph(); void saveTextFileUsingGraph(List<String> text); final Collection<String> messageList; }### Answer:
@Test public void testPrintErrors() { demo.setChannel(channel); demo.printErrors(); int count = 201; for (int i = 0; i < count; i++) { channel.publish("Error: " + i); } try { Thread.sleep(2000); } catch (Exception e) { throw new RuntimeException(e); } assertFalse(demo.messageList.isEmpty()); assertEquals(count, demo.messageList.size()); assertNotNull(demo.publisher); }