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);   }
@Test void json() { String json = "{" + "\"@type\":\"TopicMessage\"," + "\"chunk_num\":1," + "\"chunk_total\":2," + "\"consensus_timestamp\":1594401417000000000," + "\"message\":\"AQID\"," + "\"payer_account_id\":4294968296," + "\"realm_num\":0," + "\"running_hash\":\"BAUG\"," + "\"running_hash_version\":2," + "\"sequence_number\":1," + "\"topic_num\":1001," + "\"valid_start_timestamp\":1594401416000000000" + "}"; TopicMessage topicMessage = TopicMessage.builder().chunkNum(1) .chunkTotal(2) .consensusTimestamp(Instant.ofEpochSecond(1594401417)) .message(new byte[] {1, 2, 3}) .payerAccountId(4294968296L) .realmNum(0) .runningHash(new byte[] {4, 5, 6}) .runningHashVersion(2) .sequenceNumber(1L) .topicNum(1001) .validStartTimestamp(Instant.ofEpochSecond(1594401416)).build(); TopicMessageFilter filter = TopicMessageFilter.builder() .startTime(Instant.EPOCH) .topicNum(1001) .build(); topicListener.listen(filter) .as(StepVerifier::create) .thenAwait(Duration.ofMillis(50)) .then(() -> jdbcTemplate.execute("NOTIFY topic_message, '" + json + "'")) .expectNext(topicMessage) .thenCancel() .verify(Duration.ofMillis(500L)); }
@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)); }