@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(); }
@Override public void close() { writeStatusEndOfStream(); }
IpcProtocol implements IpcReader<T>, IpcWriter<T> { @Override public void close() { writeStatusEndOfStream(); } }
IpcProtocol implements IpcReader<T>, IpcWriter<T> { @Override public void close() { writeStatusEndOfStream(); } IpcProtocol(IpcBuffer buffer, EncodingFactory<T> encodingFactory); }
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); }
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);  }