コードを考えると:
logItemPublisher
.buffer(
loggingProperties.getBufferTimeoutMillis(),
TimeUnit.MILLISECONDS,
loggingProperties.getBufferSize(),
logDispatchScheduler
)
.onBackpressureDrop(droppedLogsHandler)
// persist uses Spring RestOperations
.flatMap(logs -> persist(logs, url)
.subscribeOn(logDispatchScheduler)
)
.subscribe();
どこ、
logItemPublisher = ReplaySubject.<AbstractHttpLogItem>createWithSize(4 * loggingProperties.getBufferSize())
.toSerialized();
logDispatchScheduler = new TestScheduler()
私は単体テストを持っています:
@Test
public void testLogServiceSlow() {
loggingProperties.setBufferSize(1);
// rx.ring-buffer.size property stores the size of any in-memory ring buffers that RxJava uses when an
// Observable cannot keep up with rate of event emissions.
// default value is 128 on the JVM; RxJava 2.x makes this configurable in flatMap
System.setProperty("rx.ring-buffer.size", "2");
// this is what persist does
when(restOperations.postForEntity(anyString(), any(HttpEntity.class), eq(Void.class)))
.thenAnswer(invocation -> {
Thread.sleep(500);
return ResponseEntity.ok().build();
});
logServiceClient.persistLogs(logs);
scheduler.advanceTimeBy(2L, TimeUnit.SECONDS);
System.clearProperty("rx.ring-buffer.size");
Mockito.verify(restOperations, times(2))
.postForEntity(Mockito.eq("http://log:9000/log/service"), logsCaptors.capture(), eq(Void.class));
}
Thread.sleep
にもかかわらず、背圧が生成されないため、テストは失敗します。なぜそうでないのか理解できません。内部リング バッファは 2 つのアイテムの後にいっぱいになり、残りのアイテムは削除されます。