class 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();
                    }
                }));
    }

    public  TitusClientImpl(JobManagementServiceStub jobManagementService,
                           JobManagementServiceFutureStub jobManagementServiceFutureStub,
                           Registry registry);

}