TitusClientImpl implements TitusClient { @Override public Flux<JobOrTaskUpdate> getJobAndTaskUpdates() { return Flux.create(sink -> attachCallerId(jobManagementService, CLIENT_ID) .observeJobs(ObserveJobsQuery.newBuilder().build(), new StreamObserver<JobChangeNotification>() { @Override public void onNext(JobChangeNotification jobChangeNotification) { switch (jobChangeNotification.getNotificationCase()) { case JOBUPDATE: final Job job = jobChangeNotification.getJobUpdate().getJob(); jobs.put(job.getId(), CompletableFuture.completedFuture(job)); logger.debug("<{}> JobUpdate {}", Thread.currentThread().getName(), jobChangeNotification.getJobUpdate().getJob().getId()); sink.next(JobOrTaskUpdate.jobUpdate(job)); numJobUpdates.incrementAndGet(); break; case TASKUPDATE: logger.debug("<{}> TaskUpdate {}", Thread.currentThread().getName(), jobChangeNotification.getTaskUpdate().getTask().getId()); final Task task = jobChangeNotification.getTaskUpdate().getTask(); sink.next(JobOrTaskUpdate.taskUpdate(task)); numTaskUpdates.incrementAndGet(); break; case SNAPSHOTEND: logger.info("<{}> SnapshotEnd {}", Thread.currentThread().getName(), jobChangeNotification); numSnapshotUpdates.incrementAndGet(); break; default: logger.error("<{}> Unknown Notification ? {}", Thread.currentThread().getName(), jobChangeNotification.getNotificationCase()); } } @Override public void onError(Throwable t) { logger.error("Exception in ObserveJobs :: ", t); apiErrors.incrementAndGet(); sink.error(t); } @Override public void onCompleted() { logger.info("STREAM completed ?"); sink.complete(); } })); } TitusClientImpl(JobManagementServiceStub jobManagementService,
                           JobManagementServiceFutureStub jobManagementServiceFutureStub,
                           Registry registry); @Override Mono<Task> getTask(String taskId); @Override Flux<JobOrTaskUpdate> getJobAndTaskUpdates(); @Override Mono<Job> getJobById(String jobId);  }
@Test public void getTaskUpdates() { final CountDownLatch latch = new CountDownLatch(1); final AtomicInteger tasksCount = new AtomicInteger(0); titusClient.getJobAndTaskUpdates().subscribe(task -> { if (tasksCount.incrementAndGet() == 5) { latch.countDown(); } }, e -> fail("getTaskUpdates exception {}", e)); try { latch.await(1, TimeUnit.SECONDS); } catch (InterruptedException e) { fail("getTaskUpdates Timeout ", e); } }