JobExecutorTest.java 文件源码

@Test
public void testExecuteWritesStdoutToTheStdoutListener() throws Throwable {
    final JobExecutor jobExecutor = getInstance();
    final String msgSuppliedToEcho = generateRandomString();
    final PersistedJob req =
            standardRequestWithCommand("echo", msgSuppliedToEcho);
    final AtomicReference<byte[]> bytesEchoedToStdout = new AtomicReference<>(new byte[]{});
    final Subject<byte[]> stdoutSubject = PublishSubject.create();

    stdoutSubject.subscribe(bytes ->
            bytesEchoedToStdout.getAndUpdate(existingBytes ->
                    Bytes.concat(existingBytes, bytes)));

    final Semaphore s = new Semaphore(1);
    s.acquire();
    stdoutSubject.doOnComplete(s::release).subscribe();

    final JobEventListeners listeners =
            createStdoutListener(stdoutSubject);

    jobExecutor.execute(req, listeners);

    s.tryAcquire(TestConstants.DEFAULT_TIMEOUT, MILLISECONDS);

    final String stringFromStdout = new String(bytesEchoedToStdout.get()).trim();
    assertThat(stringFromStdout).isEqualTo(msgSuppliedToEcho);
}