ServerEventRelay implements EventRelay, EventHandler { public void relayTopicEvent(MotechEvent event) { verifyEventNotNull(event); Set<EventListener> listeners = getEventListeners(event); for (EventListener listener : listeners) { handleTopicEvent(listener, event); } if (proxyInOsgi(event)) { sendInOSGi(event); } } @Autowired  ServerEventRelay(OutboundEventGateway outboundEventGateway, EventListenerRegistry eventListenerRegistry, MotechEventConfig motechEventConfig,
                            EventAdmin osgiEventAdmin, BundleContext bundleContext); @Override void sendEventMessage(MotechEvent event); @Override void broadcastEventMessage(MotechEvent event); void relayQueueEvent(MotechEvent event); void relayTopicEvent(MotechEvent event); @Override void handleEvent(Event osgiEvent);  }
@Test(expected = IllegalArgumentException.class) public void testRelayNullTopicEvent() throws Exception { eventRelay.relayTopicEvent(null); }
@Test public void shouldRetryEventHandlingWhenRelyingTopicEvent() { final BooleanValue handled = new BooleanValue(false); when(motechEventConfig.getMessageMaxRedeliveryCount()).thenReturn(2); when(eventListener.getIdentifier()).thenReturn("retrying"); doThrow(new RuntimeException()) .doThrow(new RuntimeException()) .doAnswer(new Answer<Void>() { @Override public Void answer(InvocationOnMock invocationOnMock) throws Throwable { handled.setValue(true); return null; } }) .when(eventListener).handle(any(MotechEvent.class)); setUpListeners(SUBJECT, eventListener); eventRelay.relayTopicEvent(new MotechEvent(SUBJECT)); verify(eventListener, times(3)).handle(any(MotechEvent.class)); assertTrue(handled.getValue()); verify(eventAdmin, never()).postEvent(any(Event.class)); verify(eventAdmin, never()).sendEvent(any(Event.class)); }
@Test public void shouldStopRetryingEventHandlingAfterMaxRedeliveryCountIsHitWhenRelyingTopicEvent() { final BooleanValue handled = new BooleanValue(false); when(motechEventConfig.getMessageMaxRedeliveryCount()).thenReturn(2); when(eventListener.getIdentifier()).thenReturn("retrying"); doThrow(new RuntimeException()) .doThrow(new RuntimeException()) .doThrow(new RuntimeException()) .doAnswer(new Answer<Void>() { @Override public Void answer(InvocationOnMock invocationOnMock) throws Throwable { handled.setValue(true); return null; } }) .when(eventListener).handle(any(MotechEvent.class)); setUpListeners(SUBJECT, eventListener); eventRelay.relayTopicEvent(new MotechEvent(SUBJECT)); verify(eventListener, times(3)).handle(any(MotechEvent.class)); assertFalse(handled.getValue()); }
@Test public void shouldProxyBroadcastEventsInOSGi() { Map<String, Object> params = new HashMap<>(); params.put("proxy-in-osgi", true); MotechEvent event = new MotechEvent("subject", params); eventRelay.relayTopicEvent(event); ArgumentCaptor<Event> captor = ArgumentCaptor.forClass(Event.class); verify(eventAdmin).postEvent(captor.capture()); assertEquals("subject", captor.getValue().getTopic()); }