CODEWITHSUNDEEP
javamultithreadingrx-javareactive-programmingbackpressure
Abhijit Sarkar
Abhijit Sarkar

Reputation: 24518

RxJava 1.x: How to simulate backpressure in unit test

Given the code:

logItemPublisher
    .buffer(
            loggingProperties.getBufferTimeoutMillis(),
            TimeUnit.MILLISECONDS,
            loggingProperties.getBufferSize(),
            logDispatchScheduler
    )
    .onBackpressureDrop(droppedLogsHandler)
    // persist uses Spring RestOperations
    .flatMap(logs -> persist(logs, url)
            .subscribeOn(logDispatchScheduler)
    )
    .subscribe();

where,

logItemPublisher = ReplaySubject.<AbstractHttpLogItem>createWithSize(4 * loggingProperties.getBufferSize())
                .toSerialized();
logDispatchScheduler = new TestScheduler()

I've a unit test:

@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));
}

The test fails because in spite of the Thread.sleep, no backpressure is generated. I'm failing to understand why not; the internal ring buffer should fill up after 2 items and the rest of the items should be dropped.

Upvotes: 2

Views: 590

Answers (1)

Abhijit Sarkar
Abhijit Sarkar

Reputation: 24518

Answering my own question:

int ringBufferSize = Optional.ofNullable(System.getProperty("rx.ring-buffer.size"))
    .map(Integer::parseInt)
    .orElse(128);
logItemPublisher
    .buffer(
            loggingProperties.getBufferTimeoutMillis(),
            TimeUnit.MILLISECONDS,
            loggingProperties.getBufferSize(),
            logDispatchScheduler
    )
    .onBackpressureDrop(droppedLogsHandler)
    .observeOn(logDispatchScheduler, ringBufferSize)
    .flatMap(logs -> persist(logs, url))
    .subscribe();

Then in the test, simply set ringBufferSize = 2, no need to set system property rx.ring-buffer.size.

Upvotes: 2

Related Questions