projects
/
demos
/
kafka
/
chat
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
test: HandoverIT-POC - Messages are written to 23 chat-rooms instead of 1
[demos/kafka/chat]
/
src
/
test
/
java
/
de
/
juplo
/
kafka
/
chat
/
backend
/
AbstractHandoverIT.java
diff --git
a/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java
b/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java
index
2e4faa1
..
900a7c3
100644
(file)
--- a/
src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java
+++ b/
src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java
@@
-19,6
+19,7
@@
import reactor.core.publisher.Mono;
public abstract class AbstractHandoverIT
{
static final ParameterizedTypeReference<ServerSentEvent<String>> SSE_TYPE = new ParameterizedTypeReference<>() {};
public abstract class AbstractHandoverIT
{
static final ParameterizedTypeReference<ServerSentEvent<String>> SSE_TYPE = new ParameterizedTypeReference<>() {};
+ static final int NUM_CHATROOMS = 23;
private final AbstractHandoverITContainers containers;
private final AbstractHandoverITContainers containers;
@@
-33,15
+34,21
@@
public abstract class AbstractHandoverIT
@Test
void test() throws InterruptedException
{
@Test
void test() throws InterruptedException
{
- ChatRoomInfoTo chatRoom = createChatRoom("bar").block();
+ ChatRoomInfoTo[] chatRooms = Flux
+ .range(0, NUM_CHATROOMS)
+ .flatMap(i -> createChatRoom("#" + i))
+ .toStream()
+ .toArray(size -> new ChatRoomInfoTo[size]);
+
TestClient testClient = new TestClient(
containers.haproxy.getMappedPort(8400),
TestClient testClient = new TestClient(
containers.haproxy.getMappedPort(8400),
- chatRoom,
+ chatRoom
s
,
"nerd");
testClient.run();
"nerd");
testClient.run();
- receiveMessages(chatRoom)
- .take(100)
+ Flux
+ .fromArray(chatRooms)
+ .flatMap(chatRoom ->receiveMessages(chatRoom).take(100))
.doOnNext(message -> log.info("message: {}", message))
.then()
.block();
.doOnNext(message -> log.info("message: {}", message))
.then()
.block();