EventReactor implements Service { public Subscription subscribe(Topic topic, Receiver receiver) { Objects.requireNonNull(topic); Objects.requireNonNull(receiver); Receiver theReceiver = registry.putIfAbsent(topic, receiver); if (theReceiver == null || receiver == theReceiver) { if (trace) { System.out.format("Subscribe [%s] %s\n", name, topic); } return new Subscription(topic, this); } else { if (trace) { System.out.format("Subscribe failed [%s] %s; subscription already exists for topic\n", name, topic); } return null; } } 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);  }
@Test public void testSubscribe() { Topic topic = Topics.getTopic("TestTopic1"); Subscription subscription1 = reactor.subscribe(topic, receiver); ByteBuffer src = ByteBuffer.allocate(1024); final byte[] bytes = "Hello world!".getBytes(); src.put(bytes); reactor.post(topic, src); reactor.post(topic, src); try { Thread.sleep(1000L); } catch (InterruptedException e) { } assertEquals(2, receiver.getCount()); ByteBuffer received = receiver.getLastBuffer(); byte[] dst = new byte[bytes.length]; received.get(dst); assertArrayEquals(bytes, dst); Topic topic2 = Topics.getTopic("TestTopic2"); assertTrue(topic.hashCode() != topic2.hashCode()); reactor.post(topic2, src); try { Thread.sleep(100L); } catch (InterruptedException e) { } assertEquals(2, receiver.getCount()); Subscription subscription2 = reactor.subscribe(topic2, receiver); reactor.post(topic2, src); try { Thread.sleep(100L); } catch (InterruptedException e) { } assertEquals(3, receiver.getCount()); reactor.unsubscribe(topic); reactor.post(topic, src); try { Thread.sleep(100L); } catch (InterruptedException e) { } assertEquals(3, receiver.getCount()); }