### Question:
IpcProtocol implements IpcReader<T>, IpcWriter<T> { @Override public void close() { writeStatusEndOfStream(); } IpcProtocol(IpcBuffer buffer, EncodingFactory<T> encodingFactory); void start(); @Override void send(Event<T> message); @Override void close(); @Override PollResult poll(T target);  }### Answer:
@Test public void test_concurrent_producer_and_consumer() throws Exception { Path mmf = tempDir.getRoot().toPath().resolve("mmf"); SpyListener<SuiteListener> expectations = new SpyListener<>(SuiteListener.class); lotsOfEventsForConcurrencyTesting(expectations.getListener(), 0); expectations.replay(); Runnable producer = () -> { IpcWriter<SuiteListener> writer = IpcChannel.writer(mmf, SuiteListenerEncoding::new); lotsOfEventsForConcurrencyTesting(sendTo(writer), 1); writer.close(); }; Runnable consumer = () -> { IpcReader<SuiteListener> reader = IpcChannel.reader(mmf, SuiteListenerEncoding::new); try { IpcReaders.decodeAll(reader, expectations.getListener()); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }; runConcurrently(producer, consumer); expectations.verify(); }
@Test public void producer_will_always_decide_segment_size_except_for_the_first_segment() throws Exception { Path mmf = tempDir.getRoot().toPath().resolve("mmf"); SpyListener<SuiteListener> expectations = new SpyListener<>(SuiteListener.class); smallEventsForSegmentSizeConcurrencyTesting(expectations.getListener(), 0); expectations.replay(); Runnable producer = () -> { IpcWriter<SuiteListener> writer = IpcChannel.writer(new FileSegmenter(mmf, 1, 1), SuiteListenerEncoding::new); smallEventsForSegmentSizeConcurrencyTesting(sendTo(writer), 10000000); writer.close(); }; Runnable consumer = () -> { IpcReader<SuiteListener> reader = IpcChannel.reader(new FileSegmenter(mmf, 2, 2), SuiteListenerEncoding::new); try { IpcReaders.decodeAll(reader, expectations.getListener()); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }; runConcurrently(producer, consumer); expectations.verify(); try (DirectoryStream<Path> segments = Files.newDirectoryStream(tempDir.getRoot().toPath())) { for (Path segment : segments) { if (segment.equals(mmf)) { continue; } assertThat("size of " + segment, Files.size(segment), is(1L)); } } }