@Test void jsonError() { TopicMessageFilter filter = TopicMessageFilter.builder() .startTime(Instant.EPOCH) .build(); topicListener.listen(filter) .as(StepVerifier::create) .thenAwait(Duration.ofMillis(50)) .then(() -> jdbcTemplate.execute("NOTIFY topic_message, 'invalid'")) .expectNoEvent(Duration.ofMillis(500L)) .thenCancel() .verify(Duration.ofMillis(600L)); }
private Flux<String> listen() { EmitterProcessor<String> emitterProcessor = EmitterProcessor.create(); FluxSink<String> sink = emitterProcessor.sink().onDispose(this::unlisten); channel.handler(sink::next); log.info("Listening for messages"); return emitterProcessor; }
NotifyingTopicListener extends SharedTopicListener { private Flux<String> listen() { EmitterProcessor<String> emitterProcessor = EmitterProcessor.create(); FluxSink<String> sink = emitterProcessor.sink().onDispose(this::unlisten); channel.handler(sink::next); log.info("Listening for messages"); return emitterProcessor; } }
NotifyingTopicListener extends SharedTopicListener { private Flux<String> listen() { EmitterProcessor<String> emitterProcessor = EmitterProcessor.create(); FluxSink<String> sink = emitterProcessor.sink().onDispose(this::unlisten); channel.handler(sink::next); log.info("Listening for messages"); return emitterProcessor; } NotifyingTopicListener(DbProperties dbProperties, ListenerProperties listenerProperties); }
NotifyingTopicListener extends SharedTopicListener { private Flux<String> listen() { EmitterProcessor<String> emitterProcessor = EmitterProcessor.create(); FluxSink<String> sink = emitterProcessor.sink().onDispose(this::unlisten); channel.handler(sink::next); log.info("Listening for messages"); return emitterProcessor; } NotifyingTopicListener(DbProperties dbProperties, ListenerProperties listenerProperties);  }
NotifyingTopicListener extends SharedTopicListener { private Flux<String> listen() { EmitterProcessor<String> emitterProcessor = EmitterProcessor.create(); FluxSink<String> sink = emitterProcessor.sink().onDispose(this::unlisten); channel.handler(sink::next); log.info("Listening for messages"); return emitterProcessor; } NotifyingTopicListener(DbProperties dbProperties, ListenerProperties listenerProperties);   }