projectreactor.io - junit tests identifying EventBus callback issues

ReactorEventBusCallbackTests.java

import static org.junit.Assert.assertTrue;
import static reactor.bus.selector.Selectors.$;

import org.junit.Test;

import reactor.Environment;
import reactor.bus.Event;
import reactor.bus.EventBus;

/**
 * Unit test showing unexpected {@link EventBus} behavior.
 * 
 * @author Sean Connolly
 */
public class ReactorEventBusCallbackTests {

	static {
		Environment.initialize();
	}

	/**
	 * PASSES<br>
	 * Sanity check.. does the target get notified?
	 */
	@Test
	public void shouldNotifyTargetWithoutCallback() {
		// Given
		EventBus bus = EventBus.create();
		EndPoint target = new EndPoint();
		bus.on($("target"), target::recordFired);
		// When
		bus.notify("target", Event.wrap("Hello World!"));
		// Then
		assertTrue("target should be fired", target.fired);
	}

	/**
	 * FAILS<br>
	 * Why doesn't {@code replyTo} work with {@code notify}?
	 */
	@Test
	public void shouldNotifyTargetAndReplyToCallback() {
		// Given
		EventBus bus = EventBus.create();
		EndPoint target = new EndPoint();
		EndPoint callback = new EndPoint();
		bus.on($("target"), target::recordFired);
		bus.on($("callback"), callback::recordFired);
		// When
		bus.notify("target", Event.wrap("Hello World!", "callback"));
		assertTrue("target should be fired", target.fired);
		assertTrue("callback should be fired", callback.fired);
	}

	/**
	 * PASSES<br>
	 * OK, the {@code replyTo} works with {@code send}.
	 */
	@Test
	public void shouldSendTargetAndReplyToCallback() {
		// Given
		EventBus bus = EventBus.create();
		EndPoint target = new EndPoint();
		EndPoint callback = new EndPoint();
		bus.receive($("target"), target::recordFired);
		bus.receive($("callback"), callback::recordFired);
		// When
		bus.send("target", Event.wrap("Hello World!", "callback"));
		// Then
		assertTrue("target should be fired", target.fired);
		assertTrue("callback should be fired", callback.fired);
	}

	/**
	 * PASSES<br>
	 * This is similar to the removed {@code notify} with a callback.. but what if I'm not "receiving"?
	 */
	@Test
	public void shouldSendAndReceiveTargetAndCallback() {
		// Given
		EventBus bus = EventBus.create();
		EndPoint target = new EndPoint();
		EndPoint callback = new EndPoint();
		bus.receive($("target"), target::recordFired);
		// When
		bus.sendAndReceive("target", Event.wrap("Hello World!"), callback::recordFired);
		// Then
		assertTrue("target should be fired", target.fired);
		assertTrue("callback should be fired", callback.fired);
	}

	/**
     * FAILS<br>
	 * Yes, {@code sendAndReceive} shouldn't work with {@code on()}.. but what replaces the missing {@code notify} with
	 * a callback then?
	 */
	@Test
	public void shouldNotifyAndReceiveTargetAndCallback2() {
		// Given
		EventBus bus = EventBus.create();
		EndPoint target = new EndPoint();
		EndPoint callback = new EndPoint();
		bus.on($("target"), target::recordFired);
		// When
		bus.sendAndReceive("target", Event.wrap("Hello World!"), callback::recordFired);
		// Then
		assertTrue("target should be fired", target.fired);
		assertTrue("callback should be fired", callback.fired);
	}

	/**
	 * Simple utility to facilitate testing.
	 */
	private static class EndPoint {

		private boolean fired = false;

		public Void recordFired(Event e) {
			System.out.println(e.getData());
			fired = true;
			return null;
		}

	}

}