JobsDAOTest.java 文件源码

@Test
public void testPersistStderrReturnsADisposableThatStopsFurtherReads() {
    final JobDAO dao = getInstance();
    final JobId jobId = dao.persist(STANDARD_VALID_REQUEST).getId();
    final Subject<byte[]> stderrSubject = PublishSubject.create();
    final AtomicBoolean stderrObsWasRead = new AtomicBoolean(false);
    final Observable<byte[]> stderrObs = stderrSubject.map(data -> {
        stderrObsWasRead.set(true);
        return data;
    });

    final Disposable disposable = dao.appendStderr(jobId, stderrObs);
    disposable.dispose();
    stderrSubject.onNext(TestHelpers.generateRandomBytes());

    assertThat(stderrObsWasRead.get());
}