return Mono.error(e);
}
})
+ .doOnNext(message -> log.info(
+ "Received a message from chat-room {}: {}",
+ chatRoom,
+ message))
.take(30))
.takeUntil(message -> !running)
- .subscribe(message -> log.info("Received message: {}", message));
+ .then()
+ .block();
}
Flux<ServerSentEvent<String>> receiveMessages(ChatRoomInfoTo chatRoom)