void shouldConsumeMessages()

in pulsar-client-reactive-adapter/src/intTest/java/org/apache/pulsar/reactive/client/adapter/ReactiveMessagePipelineE2ETest.java [64:89]

• 21 lines of code
• 1 McCabe index (conditional complexity)

	void shouldConsumeMessages() throws Exception {
		try (PulsarClient pulsarClient = SingletonPulsarContainer.createPulsarClient()) {
			String topicName = "test" + UUID.randomUUID();
			// create subscription to retain messages
			pulsarClient.newConsumer(Schema.STRING).topic(topicName).subscriptionName("sub").subscribe().close();

			ReactivePulsarClient reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient);

			ReactiveMessageSender<String> messageSender = reactivePulsarClient.messageSender(Schema.STRING)
					.topic(topicName).build();
			messageSender.sendMany(Flux.range(1, 100).map(Object::toString).map(MessageSpec::of)).blockLast();

			List<String> messages = Collections.synchronizedList(new ArrayList<>());
			CountDownLatch latch = new CountDownLatch(100);

			try (ReactiveMessagePipeline ignored = reactivePulsarClient.messageConsumer(Schema.STRING)
					.subscriptionName("sub").topic(topicName).build().messagePipeline()
					.messageHandler((message) -> Mono.fromRunnable(() -> {
						messages.add(message.getValue());
						latch.countDown();
					})).build().start()) {
				assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
				assertThat(messages).isEqualTo(Flux.range(1, 100).map(Object::toString).collectList().block());
			}
		}
	}