NotificationBus { protected void addHandler(Object handler) { List<Class<?>> newNotifications = collectSubscribingTypes(handler); pointToPointValidator.validate(handler, newNotifications); Lock lock = rwLock.writeLock(); try { lock.lock(); newNotifications.forEach(type -> registeredNotifications.put(type, type)); clearCache(); eventBus.register(handler); } finally { lock.unlock(); } } protected  NotificationBus(Class<N> notificationClass); void clearAllHandlers(); void broadcast(N notification); void clearThread(); @Override String toString(); static final String NOTIFICATION_BUS_ID; }
@Test public void testCallingBroadcastFromAtLeastTwoThreadsFails() throws InterruptedException { WithinAppNotificationRouter router = WithinAppNotificationRouter.get(); NotificationBus<Notification> bus = new TestBus(); router.registerExecutionLayer(new NonExecutingEventScheduler(TestSchedulerType.TEST_SCHEDULER_TYPE, TimeProvider.NULL), bus); OuterNotificationHandler outerHandler = new OuterNotificationHandler(); router.addHandler(outerHandler); int numberOfBogusThreads = 10; List<LocalBroadcastRunnable> notifications = new ArrayList<>(); try { for (int i = 0; i < numberOfBogusThreads + 1; i++) { LocalBroadcastRunnable r = new LocalBroadcastRunnable(router); notifications.add(r); new Thread(r).start(); } for (LocalBroadcastRunnable r : notifications) { r.waitForRunning(); } Thread.sleep(50); for (LocalBroadcastRunnable r : notifications) { r.stop = true; } } finally { boolean success = false; int exceptionCount = 0; for (LocalBroadcastRunnable r : notifications) { Throwable t = r.waitForCompletion(); if (t != null) { ++exceptionCount; Assertions.assertTrue(t instanceof IllegalStateException); } else { Assertions.assertFalse(success, "At most one thread should succeed"); success = true; } } Assertions.assertTrue(success, "At least one thread should succeed"); Assertions.assertEquals(numberOfBogusThreads, exceptionCount, "All bogus threads should have failed"); } }
@Test public void testWhenTwoThreadsBroadcast_thenTheFirstAlwaysSucceedsAndTheSecondAlwaysFails() throws InterruptedException { WithinAppNotificationRouter router = WithinAppNotificationRouter.get(); NotificationBus<Notification> bus = new TestBus(); router.registerExecutionLayer(new NonExecutingEventScheduler(TestSchedulerType.TEST_SCHEDULER_TYPE, TimeProvider.NULL), bus); OuterNotificationHandler outerHandler = new OuterNotificationHandler(); router.addHandler(outerHandler); TwoThreadTestBroadcaster firstBroadcaster = new TwoThreadTestBroadcaster(router); new Thread(firstBroadcaster::runExpectingSuccess).start(); firstBroadcaster.waitForCount(10); TwoThreadTestBroadcaster secondBroadcaster = new TwoThreadTestBroadcaster(router); new Thread(secondBroadcaster::runExpectingFailure).start();; secondBroadcaster.waitForCount(100); firstBroadcaster.waitForCount(firstBroadcaster.broadcastCount + 100); firstBroadcaster.waitToStop(); secondBroadcaster.waitToStop(); Assertions.assertFalse(firstBroadcaster.failed, "Should never throw as first caller"); Assertions.assertFalse(secondBroadcaster.failed, "Should never succeed as second caller"); }
@Test void testThrowsIllegalArgument_whenNotificationMissedFromSubscriber() { NotificationBus<Notification> bus = new TestBus(); assertThatThrownBy(() -> bus.addHandler(new Subscriber() { @Subscribe public void invalidHandler() { } @Override public EventSchedulerType getSchedulerType() { return TestSchedulerType.TEST_SCHEDULER_TYPE; } })) .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("invalidHandler"); }
@Test void testThrowsIllegalArgument_whenSubscriberParameterNotANotification() { NotificationBus<Notification> bus = new TestBus(); assertThatThrownBy(() -> bus.addHandler(new Subscriber() { @Subscribe public void invalidHandler(String notANotification) { } @Override public EventSchedulerType getSchedulerType() { return TestSchedulerType.TEST_SCHEDULER_TYPE; } })) .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("invalidHandler") .hasMessageContaining("String"); }