@Test public void future() throws InterruptedException, ExecutionException, TimeoutException { Topic topic = Topics.getTopic("TestTopic2"); EventFuture future = new EventFuture(topic, reactor); Thread thread = new Thread(new Runnable() { public void run() { ByteBuffer src = ByteBuffer.allocate(1024); src.put("Hello, world".getBytes()); reactor.post(topic, src); } }); thread.start(); ByteBuffer result = future.get(1000L, TimeUnit.MILLISECONDS); assertTrue(future.isDone() && !future.isCancelled()); }
public void post(Topic topic, T src) { long sequence = ringBuffer.next(); final BufferEvent event = ringBuffer.get(sequence); event.set(topic, src); ringBuffer.publish(sequence); }
EventReactor implements Service { public void post(Topic topic, T src) { long sequence = ringBuffer.next(); final BufferEvent event = ringBuffer.get(sequence); event.set(topic, src); ringBuffer.publish(sequence); } }
EventReactor implements Service { public void post(Topic topic, T src) { long sequence = ringBuffer.next(); final BufferEvent event = ringBuffer.get(sequence); event.set(topic, src); ringBuffer.publish(sequence); } protected  EventReactor(Builder<T, ?, ?> builder); }
EventReactor implements Service { public void post(Topic topic, T src) { long sequence = ringBuffer.next(); final BufferEvent event = ringBuffer.get(sequence); event.set(topic, src); ringBuffer.publish(sequence); } protected  EventReactor(Builder<T, ?, ?> builder); @SuppressWarnings("rawtypes") static Builder builder(); void close(); boolean hasSubscriber(Topic topic); @SuppressWarnings("unchecked") CompletableFuture<? extends EventReactor<T>> open(); void post(Topic topic, T src); TimerSchedule postAt(Topic topic, T src, Date time); TimerSchedule postAt(Topic topic, T src, long delayMillis); TimerSchedule postAtInterval(Topic topic, T src, long periodMillis); void setTrace(boolean trace); void setTrace(boolean trace, String name); Subscription subscribe(Topic topic, Receiver receiver); void unsubscribe(Topic topic); }
EventReactor implements Service { public void post(Topic topic, T src) { long sequence = ringBuffer.next(); final BufferEvent event = ringBuffer.get(sequence); event.set(topic, src); ringBuffer.publish(sequence); } protected  EventReactor(Builder<T, ?, ?> builder); @SuppressWarnings("rawtypes") static Builder builder(); void close(); boolean hasSubscriber(Topic topic); @SuppressWarnings("unchecked") CompletableFuture<? extends EventReactor<T>> open(); void post(Topic topic, T src); TimerSchedule postAt(Topic topic, T src, Date time); TimerSchedule postAt(Topic topic, T src, long delayMillis); TimerSchedule postAtInterval(Topic topic, T src, long periodMillis); void setTrace(boolean trace); void setTrace(boolean trace, String name); Subscription subscribe(Topic topic, Receiver receiver); void unsubscribe(Topic topic);  }