@Test void testReactiveStream() { when(mockEvent.getObject()).thenReturn(of(rdf.createIRI(TRELLIS_DATA_PREFIX + "resource"))); when(mockEvent.getAgents()).thenReturn(singleton(Trellis.AdministratorAgent)); when(mockEvent.getIdentifier()).thenReturn(rdf.createIRI("urn:test")); when(mockEvent.getCreated()).thenReturn(time); when(mockEvent.getTypes()).thenReturn(singleton(AS.Update)); when(mockEvent.getObjectTypes()).thenReturn(singleton(LDP.RDFSource)); when(mockEvent.getInbox()).thenReturn(empty()); service.emit(mockEvent); await().atMost(5, SECONDS).until(() -> collector.getResults().size() == 1); assertEquals(1, collector.getResults().size(), "Incorrect number of messages!"); service.emit(mockEvent); await().atMost(5, SECONDS).until(() -> collector.getResults().size() == 2); assertEquals(2, collector.getResults().size(), "Incorrect number of messages!"); service.emit(mockEvent); service.emit(mockEvent); await().atMost(5, SECONDS).until(() -> collector.getResults().size() == 4); assertEquals(4, collector.getResults().size(), "Incorrect number of messages!"); }
@Override public void emit(final Event event) { LOGGER.debug("Sending message to reactive destination: {}", event.getIdentifier()); trellisEvent.fireAsync(event); subject.onNext(of(serializer.serialize(event))); }
ReactiveEventService implements EventService { @Override public void emit(final Event event) { LOGGER.debug("Sending message to reactive destination: {}", event.getIdentifier()); trellisEvent.fireAsync(event); subject.onNext(of(serializer.serialize(event))); } }
ReactiveEventService implements EventService { @Override public void emit(final Event event) { LOGGER.debug("Sending message to reactive destination: {}", event.getIdentifier()); trellisEvent.fireAsync(event); subject.onNext(of(serializer.serialize(event))); }  }
ReactiveEventService implements EventService { @Override public void emit(final Event event) { LOGGER.debug("Sending message to reactive destination: {}", event.getIdentifier()); trellisEvent.fireAsync(event); subject.onNext(of(serializer.serialize(event))); }  @Override void emit(final Event event); @Outgoing(REACTIVE_DESTINATION) Flowable<Message<String>> publish(); }
ReactiveEventService implements EventService { @Override public void emit(final Event event) { LOGGER.debug("Sending message to reactive destination: {}", event.getIdentifier()); trellisEvent.fireAsync(event); subject.onNext(of(serializer.serialize(event))); }  @Override void emit(final Event event); @Outgoing(REACTIVE_DESTINATION) Flowable<Message<String>> publish(); static final String REACTIVE_DESTINATION; }