@Test public void subscriptionCallbackAppliesBackpressureInCategoryPublisher() throws IOException { when(reader.readValue((byte[])any())).thenReturn(new Object()); Publisher<EventSubscriptionUpdate<Event>> publisher = repository.getPublisher(null); CatchUpSubscription catchUpSubscription = mock(CatchUpSubscription.class); when(eventStore.subscribeToStreamFrom( anyString(), any(Long.class), any(CatchUpSubscriptionSettings.class), any(CatchUpSubscriptionListener.class), any(UserCredentials.class))) .thenReturn(catchUpSubscription); AtomicInteger seenEvents = new AtomicInteger(0); publisher.subscribe(Subscribers.bounded(100, event -> { seenEvents.incrementAndGet(); })); ArgumentCaptor<CatchUpSubscriptionListener> listenerCaptor = ArgumentCaptor.forClass(CatchUpSubscriptionListener.class); verify(eventStore, times(1)).subscribeToStreamFrom( eq("$ce-pref"), eq(null), any(CatchUpSubscriptionSettings.class), listenerCaptor.capture()); CatchUpSubscriptionListener listener = listenerCaptor.getValue(); Assert.assertEquals(0, seenEvents.get()); listener.onEvent( catchUpSubscription, new ResolvedEvent( EventStoreClientMessages .ResolvedEvent .newBuilder() .setEvent(EventStoreClientMessages.EventRecord.newBuilder() .setCreatedEpoch(2342354) .setEventStreamId("stream") .setEventId(ByteString.copyFrom( UUIDConverter.toBytes(UUID.randomUUID()))) .setEventNumber(42) .setEventType("type") .setDataContentType(0) .setMetadataContentType(0) .setData(ByteString.copyFrom("{}".getBytes()))) .setCommitPosition(0) .setPreparePosition(0) .build())); Assert.assertEquals(1, seenEvents.get()); }
@Override public Publisher<EventSubscriptionUpdate<T>> getPublisher(final Integer fromVersion) { logger.info("Creating publisher for all events in {} (starting with version {})", streamPrefix, fromVersion); return Flux.create(emitter -> { final CatchUpSubscription subscription = eventStore.subscribeToStreamFrom( getCategoryStreamName(), convertTo64Bit(fromVersion), defaultSubscriptionSettings, new EmitterListener(emitter, streamPrefix + "-all")); emitter.setCancellation(() -> { logger.info("Closing ESJC subscription (asynchronously)"); subscription.stop(); }); }); }
EventStoreEsjcEventRepository implements EventRepository<T> { @Override public Publisher<EventSubscriptionUpdate<T>> getPublisher(final Integer fromVersion) { logger.info("Creating publisher for all events in {} (starting with version {})", streamPrefix, fromVersion); return Flux.create(emitter -> { final CatchUpSubscription subscription = eventStore.subscribeToStreamFrom( getCategoryStreamName(), convertTo64Bit(fromVersion), defaultSubscriptionSettings, new EmitterListener(emitter, streamPrefix + "-all")); emitter.setCancellation(() -> { logger.info("Closing ESJC subscription (asynchronously)"); subscription.stop(); }); }); } }
EventStoreEsjcEventRepository implements EventRepository<T> { @Override public Publisher<EventSubscriptionUpdate<T>> getPublisher(final Integer fromVersion) { logger.info("Creating publisher for all events in {} (starting with version {})", streamPrefix, fromVersion); return Flux.create(emitter -> { final CatchUpSubscription subscription = eventStore.subscribeToStreamFrom( getCategoryStreamName(), convertTo64Bit(fromVersion), defaultSubscriptionSettings, new EmitterListener(emitter, streamPrefix + "-all")); emitter.setCancellation(() -> { logger.info("Closing ESJC subscription (asynchronously)"); subscription.stop(); }); }); } EventStoreEsjcEventRepository(
            final String streamPrefix,
            final EventStore eventStore,
            final Class<T> eventClass,
            final ObjectMapper objectMapper,
            final EventNormalizer<T> normalizer); }
EventStoreEsjcEventRepository implements EventRepository<T> { @Override public Publisher<EventSubscriptionUpdate<T>> getPublisher(final Integer fromVersion) { logger.info("Creating publisher for all events in {} (starting with version {})", streamPrefix, fromVersion); return Flux.create(emitter -> { final CatchUpSubscription subscription = eventStore.subscribeToStreamFrom( getCategoryStreamName(), convertTo64Bit(fromVersion), defaultSubscriptionSettings, new EmitterListener(emitter, streamPrefix + "-all")); emitter.setCancellation(() -> { logger.info("Closing ESJC subscription (asynchronously)"); subscription.stop(); }); }); } EventStoreEsjcEventRepository(
            final String streamPrefix,
            final EventStore eventStore,
            final Class<T> eventClass,
            final ObjectMapper objectMapper,
            final EventNormalizer<T> normalizer); @Override Class<T> getEventType(); @Override EventReadResult<T> readAll(final int version, final int maxEvents); @Override EventReadResult<T> read(final String streamId, final int version, final int maxEvents); @Override EventRecord<T> readFirst(final String streamId); @Override EventRecord<T> readLast(final String streamId); @Override int append(
            final String streamId,
            final List<EventData<T>> events,
            final ExpectedVersion version); @Override int getCurrentVersion(); @Override int getCurrentVersion(final String streamId); @Override Publisher<EventSubscriptionUpdate<T>> getStreamPublisher(
            final String streamId,
            final Integer fromVersion); @Override Publisher<EventSubscriptionUpdate<T>> getPublisher(final Integer fromVersion); }
EventStoreEsjcEventRepository implements EventRepository<T> { @Override public Publisher<EventSubscriptionUpdate<T>> getPublisher(final Integer fromVersion) { logger.info("Creating publisher for all events in {} (starting with version {})", streamPrefix, fromVersion); return Flux.create(emitter -> { final CatchUpSubscription subscription = eventStore.subscribeToStreamFrom( getCategoryStreamName(), convertTo64Bit(fromVersion), defaultSubscriptionSettings, new EmitterListener(emitter, streamPrefix + "-all")); emitter.setCancellation(() -> { logger.info("Closing ESJC subscription (asynchronously)"); subscription.stop(); }); }); } EventStoreEsjcEventRepository(
            final String streamPrefix,
            final EventStore eventStore,
            final Class<T> eventClass,
            final ObjectMapper objectMapper,
            final EventNormalizer<T> normalizer); @Override Class<T> getEventType(); @Override EventReadResult<T> readAll(final int version, final int maxEvents); @Override EventReadResult<T> read(final String streamId, final int version, final int maxEvents); @Override EventRecord<T> readFirst(final String streamId); @Override EventRecord<T> readLast(final String streamId); @Override int append(
            final String streamId,
            final List<EventData<T>> events,
            final ExpectedVersion version); @Override int getCurrentVersion(); @Override int getCurrentVersion(final String streamId); @Override Publisher<EventSubscriptionUpdate<T>> getStreamPublisher(
            final String streamId,
            final Integer fromVersion); @Override Publisher<EventSubscriptionUpdate<T>> getPublisher(final Integer fromVersion);  }