void shouldReadMessages()

in pulsar-client-reactive-adapter/src/intTest/java/org/apache/pulsar/reactive/client/adapter/ReactiveMessageReaderE2ETest.java [41:60]

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

	void shouldReadMessages() throws Exception {
		try (PulsarClient pulsarClient = SingletonPulsarContainer.createPulsarClient();
				ReactiveMessageSenderCache producerCache = AdaptedReactivePulsarClientFactory.createCache()) {
			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)
					.cache(producerCache).topic(topicName).build();
			messageSender.sendMany(Flux.range(1, 100).map(Object::toString).map(MessageSpec::of)).blockLast();

			ReactiveMessageReader<String> messageReader = reactivePulsarClient.messageReader(Schema.STRING)
					.topic(topicName).build();
			List<String> messages = messageReader.readMany().map(Message::getValue).collectList().block();

			assertThat(messages).isEqualTo(Flux.range(1, 100).map(Object::toString).collectList().block());
		}
	}