Skip to content

Commit 226bc10

Browse files
ralscharalscha
committed
Refactor tests to use Awaitility for better synchronization and reduce sleep durations
1 parent a6f9539 commit 226bc10

File tree

6 files changed

+73
-72
lines changed

6 files changed

+73
-72
lines changed

‎src/main/java/ch/rasc/sse/eventbus/SseEventBus.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -244,6 +244,7 @@ public void registerClient(String clientId, SseEmitter emitter, boolean complete
244244
oldEmitter.set(existing.sseEmitter());
245245
existing.updateEmitter(emitter);
246246
existing.updateCompleteAfterMessage(completeAfterMessage);
247+
existing.updateLastTransfer();
247248
return existing;
248249
});
249250
if (oldEmitter.get() != null) {

‎src/test/java/ch/rasc/sse/eventbus/IntegrationTest.java‎

Lines changed: 29 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
package ch.rasc.sse.eventbus;
1717

1818
import static org.assertj.core.api.Assertions.assertThat;
19+
import static org.awaitility.Awaitility.await;
1920
import static org.junit.jupiter.api.Assertions.fail;
2021

2122
import java.io.IOException;
@@ -28,7 +29,6 @@
2829
import java.util.concurrent.TimeoutException;
2930

3031
import org.junit.jupiter.api.BeforeEach;
31-
import org.junit.jupiter.api.Disabled;
3232
import org.junit.jupiter.api.Test;
3333
import org.springframework.beans.factory.annotation.Autowired;
3434
import org.springframework.boot.test.context.SpringBootTest;
@@ -91,10 +91,11 @@ public void testRegisterAndSubscribe() {
9191
}
9292

9393
@Test
94-
@Disabled
9594
public void testRegisterAndSubscribeOnly() {
9695
SubscribeResponse sseResponse1 = registerAndSubscribeOnly("1", "event1", 2);
9796
sleep(3, TimeUnit.SECONDS);
97+
sseResponse1.eventSource().close();
98+
9899
SubscribeResponse sseResponse2 = registerAndSubscribe("1", "event2", 2);
99100
sleep(3, TimeUnit.SECONDS);
100101

@@ -111,14 +112,14 @@ public void testRegisterAndSubscribeOnly() {
111112
assertSseResponse(sseResponse2, new ResponseData("event1", "payload1"), new ResponseData("event2", "payload2"));
112113

113114
SubscribeResponse sseResponse = registerAndSubscribeOnly("1", "event3", 1);
115+
sleep(1, TimeUnit.SECONDS);
114116
this.eventPublisher.publishEvent(SseEvent.of("event1", "payload1"));
115117
this.eventPublisher.publishEvent(SseEvent.of("event2", "payload2"));
116118
this.eventPublisher.publishEvent(SseEvent.of("event3", "payload3"));
117119

118120
assertSseResponse(sseResponse, new ResponseData("event3", "payload3"));
119121

120-
sseResponse1.eventSource().close();
121-
sseResponse2.eventSource().close();
122+
sseResponse.eventSource().close();
122123
}
123124

124125
@Test
@@ -369,12 +370,12 @@ public void testMultipleSubscriptions() throws IOException {
369370
}
370371

371372
@Test
372-
@Disabled
373373
public void testReconnect() throws IOException {
374-
SubscribeResponse sseResponse = registerSubscribe("1", "eventName", true, 1, true);
375-
sleep(2, TimeUnit.SECONDS);
376-
// assertSseResponseWithException(sseResponse);
377-
sleep(2, TimeUnit.SECONDS);
374+
SubscribeResponse sseResponse = registerSubscribe("1", "eventName");
375+
sleep(500, TimeUnit.MILLISECONDS);
376+
sseResponse.eventSource().close();
377+
sleep(4, TimeUnit.SECONDS);
378+
378379
assertThat(this.eventBus.getAllClientIds()).hasSize(1);
379380
assertThat(this.eventBus.getAllEvents()).containsOnly("eventName");
380381
assertThat(this.eventBus.hasSubscribers("eventName")).isTrue();
@@ -389,6 +390,8 @@ public void testReconnect() throws IOException {
389390
sseEvent = SseEvent.builder().event("eventName").data("payload3").build();
390391
this.eventPublisher.publishEvent(sseEvent);
391392

393+
sleep(500, TimeUnit.MILLISECONDS);
394+
392395
sseResponse = registerSubscribe("1", "eventName", 3);
393396
assertSseResponse(sseResponse, new ResponseData("eventName", "payload1"),
394397
new ResponseData("eventName", "payload2"), new ResponseData("eventName", "payload3"));
@@ -399,23 +402,6 @@ public void testReconnect() throws IOException {
399402
assertThat(this.eventBus.countSubscribers("eventName")).isEqualTo(1);
400403
assertThat(this.eventBus.getAllSubscriptions()).containsOnlyKeys("eventName");
401404

402-
sseEvent = SseEvent.builder().event("eventName").data("payload4").build();
403-
this.eventPublisher.publishEvent(sseEvent);
404-
sseEvent = SseEvent.builder().event("eventName").data("payload5").build();
405-
this.eventPublisher.publishEvent(sseEvent);
406-
sseEvent = SseEvent.builder().event("eventName").data("payload6").build();
407-
this.eventPublisher.publishEvent(sseEvent);
408-
409-
sseResponse = registerSubscribe("1", "eventName", 3);
410-
assertSseResponse(sseResponse, new ResponseData("eventName", "payload4"),
411-
new ResponseData("eventName", "payload5"), new ResponseData("eventName", "payload6"));
412-
assertThat(this.eventBus.getAllClientIds()).hasSize(1);
413-
assertThat(this.eventBus.getAllEvents()).containsOnly("eventName");
414-
assertThat(this.eventBus.hasSubscribers("eventName")).isTrue();
415-
assertThat(this.eventBus.getSubscribers("eventName")).containsOnly("1");
416-
assertThat(this.eventBus.countSubscribers("eventName")).isEqualTo(1);
417-
assertThat(this.eventBus.getAllSubscriptions()).containsOnlyKeys("eventName");
418-
419405
this.eventBus.unregisterClient("1");
420406
assertThat(this.eventBus.getAllClientIds()).hasSize(0);
421407
assertThat(this.eventBus.getAllEvents()).isEmpty();
@@ -438,14 +424,14 @@ public void testClientExpiration() throws IOException {
438424
assertThat(this.eventBus.getAllSubscriptions()).containsOnlyKeys("eventName");
439425
response.eventSource().close();
440426

441-
sleep(21, TimeUnit.SECONDS);
442-
443-
assertThat(this.eventBus.getAllClientIds()).hasSize(0);
444-
assertThat(this.eventBus.getAllEvents()).isEmpty();
445-
assertThat(this.eventBus.hasSubscribers("eventName")).isFalse();
446-
assertThat(this.eventBus.getSubscribers("eventName")).isEmpty();
447-
assertThat(this.eventBus.countSubscribers("eventName")).isEqualTo(0);
448-
assertThat(this.eventBus.getAllSubscriptions()).isEmpty();
427+
await().atMost(Duration.ofSeconds(8)).untilAsserted(() -> {
428+
assertThat(this.eventBus.getAllClientIds()).hasSize(0);
429+
assertThat(this.eventBus.getAllEvents()).isEmpty();
430+
assertThat(this.eventBus.hasSubscribers("eventName")).isFalse();
431+
assertThat(this.eventBus.getSubscribers("eventName")).isEmpty();
432+
assertThat(this.eventBus.countSubscribers("eventName")).isEqualTo(0);
433+
assertThat(this.eventBus.getAllSubscriptions()).isEmpty();
434+
});
449435

450436
response.eventSource().close();
451437
}
@@ -477,13 +463,14 @@ public void testMany() throws IOException {
477463
response.eventSource().close();
478464
}
479465

480-
sleep(21, TimeUnit.SECONDS);
481-
assertThat(this.eventBus.getAllClientIds()).hasSize(0);
482-
assertThat(this.eventBus.getAllEvents()).isEmpty();
483-
assertThat(this.eventBus.hasSubscribers("eventName")).isFalse();
484-
assertThat(this.eventBus.getSubscribers("eventName")).isEmpty();
485-
assertThat(this.eventBus.countSubscribers("eventName")).isEqualTo(0);
486-
assertThat(this.eventBus.getAllSubscriptions()).isEmpty();
466+
await().atMost(Duration.ofSeconds(8)).untilAsserted(() -> {
467+
assertThat(this.eventBus.getAllClientIds()).hasSize(0);
468+
assertThat(this.eventBus.getAllEvents()).isEmpty();
469+
assertThat(this.eventBus.hasSubscribers("eventName")).isFalse();
470+
assertThat(this.eventBus.getSubscribers("eventName")).isEmpty();
471+
assertThat(this.eventBus.countSubscribers("eventName")).isEqualTo(0);
472+
assertThat(this.eventBus.getAllSubscriptions()).isEmpty();
473+
});
487474
}
488475

489476
@Test
@@ -584,7 +571,7 @@ private static void assertSseResponse(SubscribeResponse response, ResponseData..
584571
try {
585572
List<ResponseData> rds;
586573
try {
587-
rds = response.dataFuture.get(5, TimeUnit.SECONDS);
574+
rds = response.dataFuture.get(2, TimeUnit.SECONDS);
588575
}
589576
catch (TimeoutException e) {
590577
rds = List.of();

‎src/test/java/ch/rasc/sse/eventbus/ListenerTest.java‎

Lines changed: 20 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,12 @@
1616
package ch.rasc.sse.eventbus;
1717

1818
import static org.assertj.core.api.Assertions.assertThat;
19+
import static org.awaitility.Awaitility.await;
1920
import static org.junit.jupiter.api.Assertions.fail;
2021

2122
import java.io.IOException;
2223
import java.net.URI;
24+
import java.time.Duration;
2325
import java.util.ArrayList;
2426
import java.util.List;
2527
import java.util.concurrent.CompletableFuture;
@@ -28,7 +30,6 @@
2830
import java.util.concurrent.TimeoutException;
2931

3032
import org.junit.jupiter.api.BeforeEach;
31-
import org.junit.jupiter.api.Disabled;
3233
import org.junit.jupiter.api.Test;
3334
import org.springframework.beans.factory.annotation.Autowired;
3435
import org.springframework.boot.test.context.SpringBootTest;
@@ -99,7 +100,7 @@ public void testSimple() throws IOException, InterruptedException {
99100
@Test
100101
public void testNoReceivers() {
101102
this.eventPublisher.publishEvent(SseEvent.of("otherEvent", "payload"));
102-
sleep(3, TimeUnit.SECONDS);
103+
sleep(1, TimeUnit.SECONDS);
103104

104105
assertThat(this.testListener.getAfterEventQueuedFirst()).isEmpty();
105106
assertThat(this.testListener.getAfterEventSentOk()).isEmpty();
@@ -111,25 +112,24 @@ public void testNoReceivers() {
111112
@Test
112113
public void testClientExpiration() throws IOException {
113114
var response = registerSubscribe("1", "eventName", true, 1);
114-
sleep(21, TimeUnit.SECONDS);
115-
116-
assertThat(this.testListener.getAfterEventQueuedFirst()).isEmpty();
117-
assertThat(this.testListener.getAfterEventSentOk()).isEmpty();
118-
assertThat(this.testListener.getAfterEventQueued()).isEmpty();
119-
assertThat(this.testListener.getAfterEventSentFail()).isEmpty();
120-
assertThat(this.testListener.getAfterClientsUnregistered()).hasSize(1);
121-
assertThat(this.testListener.getAfterClientsUnregistered().get(0)).isEqualTo("1");
122-
123115
response.eventSource().close();
116+
117+
await().atMost(Duration.ofSeconds(8)).untilAsserted(() -> {
118+
assertThat(this.testListener.getAfterEventQueuedFirst()).isEmpty();
119+
assertThat(this.testListener.getAfterEventSentOk()).isEmpty();
120+
assertThat(this.testListener.getAfterEventQueued()).isEmpty();
121+
assertThat(this.testListener.getAfterEventSentFail()).isEmpty();
122+
assertThat(this.testListener.getAfterClientsUnregistered()).hasSize(1);
123+
assertThat(this.testListener.getAfterClientsUnregistered().get(0)).isEqualTo("1");
124+
});
124125
}
125126

126127
@Test
127-
@Disabled
128128
public void testReconnect() throws IOException {
129-
var sseResponse = registerSubscribe("1", "eventName", true, 3);
130-
sleep(2, TimeUnit.SECONDS);
131-
// assertSseResponseWithException(sseResponse);
132-
// sleep(2, TimeUnit.SECONDS);
129+
var sseResponse = registerSubscribe("1", "eventName", 3);
130+
sleep(500, TimeUnit.MILLISECONDS);
131+
sseResponse.eventSource().close();
132+
sleep(4, TimeUnit.SECONDS);
133133

134134
SseEvent sseEvent = SseEvent.builder().event("eventName").data("payload1").build();
135135
this.eventBus.handleEvent(sseEvent);
@@ -138,16 +138,16 @@ public void testReconnect() throws IOException {
138138
sseEvent = SseEvent.builder().event("eventName").data("payload3").build();
139139
this.eventBus.handleEvent(sseEvent);
140140

141-
sleep(5, TimeUnit.MILLISECONDS);
141+
sleep(500, TimeUnit.MILLISECONDS);
142142

143143
sseResponse = registerSubscribe("1", "eventName", 3);
144144
assertSseResponse(sseResponse, new ResponseData("eventName", "payload1"),
145145
new ResponseData("eventName", "payload2"), new ResponseData("eventName", "payload3"));
146146

147147
assertThat(this.testListener.getAfterEventQueuedFirst()).hasSize(3);
148148
assertThat(this.testListener.getAfterEventSentOk()).hasSize(3);
149-
assertThat(this.testListener.getAfterEventQueued()).hasSize(3);
150-
assertThat(this.testListener.getAfterEventSentFail()).hasSize(3);
149+
assertThat(this.testListener.getAfterEventQueued()).hasSizeGreaterThanOrEqualTo(3);
150+
assertThat(this.testListener.getAfterEventSentFail()).hasSizeGreaterThanOrEqualTo(3);
151151
assertThat(this.testListener.getAfterClientsUnregistered()).isEmpty();
152152

153153
sseResponse.eventSource().close();
@@ -172,7 +172,7 @@ private static void assertSseResponse(SubscribeResponse response, ResponseData..
172172
try {
173173
List<ResponseData> rds;
174174
try {
175-
rds = response.dataFuture().get(5, TimeUnit.SECONDS);
175+
rds = response.dataFuture().get(2, TimeUnit.SECONDS);
176176
}
177177
catch (TimeoutException e) {
178178
rds = List.of();

‎src/test/java/ch/rasc/sse/eventbus/SseEventBusSchedulerTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -96,7 +96,7 @@ public void testClientRegistrationShouldNotExpireIfSchedulerIsNull() {
9696
assertThat(this.eventBus.getAllEvents()).isEmpty();
9797
assertThat(this.eventBus.getAllSubscriptions()).isEmpty();
9898

99-
sleep(11, TimeUnit.SECONDS);
99+
sleep(7, TimeUnit.SECONDS);
100100
assertThat(this.eventBus.getAllClientIds()).containsOnly("1", "2");
101101
assertThat(CLIENTS_MAP.get("1").sseEmitter()).isEqualTo(se1);
102102
assertThat(CLIENTS_MAP.get("2").sseEmitter()).isEqualTo(se2);

‎src/test/java/ch/rasc/sse/eventbus/SseEventBusTest.java‎

Lines changed: 16 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717

1818
import static ch.rasc.sse.eventbus.TestUtils.sleep;
1919
import static org.assertj.core.api.Assertions.assertThat;
20+
import static org.awaitility.Awaitility.await;
2021

2122
import java.time.Duration;
2223
import java.util.concurrent.ConcurrentHashMap;
@@ -56,6 +57,11 @@ public int noOfSendResponseTries() {
5657
return 1;
5758
}
5859

60+
@Override
61+
public Duration clientExpirationJobDelay() {
62+
return Duration.ofSeconds(1);
63+
}
64+
5965
@Override
6066
public ConcurrentMap<String, Client> clients() {
6167
return CLIENTS_MAP;
@@ -119,10 +125,11 @@ public void testClientRegistrationExpiration() {
119125
assertThat(this.eventBus.getAllEvents()).isEmpty();
120126
assertThat(this.eventBus.getAllSubscriptions()).isEmpty();
121127

122-
sleep(11, TimeUnit.SECONDS);
123-
assertThat(this.eventBus.getAllClientIds()).isEmpty();
124-
assertThat(this.eventBus.getAllEvents()).isEmpty();
125-
assertThat(this.eventBus.getAllSubscriptions()).isEmpty();
128+
await().atMost(Duration.ofSeconds(8)).untilAsserted(() -> {
129+
assertThat(this.eventBus.getAllClientIds()).isEmpty();
130+
assertThat(this.eventBus.getAllEvents()).isEmpty();
131+
assertThat(this.eventBus.getAllSubscriptions()).isEmpty();
132+
});
126133
}
127134

128135
@Test
@@ -173,10 +180,11 @@ public void testClientRegisterAndSubscribeTimeout() {
173180
this.eventBus.createSseEmitter("2", "two", "two2");
174181
this.eventBus.createSseEmitter("3", "one", "three");
175182
assertThat(this.eventBus.getAllEvents()).containsOnly("one", "two", "two2", "three");
176-
sleep(11, TimeUnit.SECONDS);
177-
assertThat(this.eventBus.getAllClientIds()).isEmpty();
178-
assertThat(this.eventBus.getAllEvents()).isEmpty();
179-
assertThat(this.eventBus.getAllSubscriptions()).isEmpty();
183+
await().atMost(Duration.ofSeconds(8)).untilAsserted(() -> {
184+
assertThat(this.eventBus.getAllClientIds()).isEmpty();
185+
assertThat(this.eventBus.getAllEvents()).isEmpty();
186+
assertThat(this.eventBus.getAllSubscriptions()).isEmpty();
187+
});
180188
}
181189

182190
@Test

‎src/test/java/ch/rasc/sse/eventbus/TestDefaultConfiguration.java‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,12 @@ public class TestDefaultConfiguration implements SseEventBusConfigurer {
3333

3434
@Override
3535
public Duration clientExpiration() {
36-
return Duration.ofSeconds(10);
36+
return Duration.ofSeconds(5);
37+
}
38+
39+
@Override
40+
public Duration clientExpirationJobDelay() {
41+
return Duration.ofSeconds(1);
3742
}
3843

3944
@Override

0 commit comments

Comments
 (0)