src/main/java/io/mantisrx/api/services/JobDiscoveryService.java
private void init() {
            if (!inited.getAndSet(true)) {
                Observable<JobSchedulingInfo> jobSchedulingInfoObs;
                switch (lookupKey.getLookupType()) {
                    case JOB_ID:
                        jobSchedulingInfoObs = mantisClient.getSchedulingChanges(lookupKey.getId());
                        break;
                    case JOB_CLUSTER:
                        jobSchedulingInfoObs = mantisClient.jobClusterDiscoveryInfoStream(lookupKey.getId());
                        break;
                    default:
                        throw new IllegalArgumentException("lookup key type is not supported " + lookupKey.getLookupType());
                }
                subscription = jobSchedulingInfoObs
                        .retryWhen(Util.getRetryFunc(log, "job scheduling info for (" + lookupKey.getLookupType() + ") " + lookupKey.id))
                        .doOnError((t) -> {
                            log.info("cleanup jobDiscoveryInfo onError for {}", lookupKey);
                            discoveryInfoBehaviorSubject.toSerialized().onError(t);
                            doOnZeroConnections.call(lookupKey);
                        })
                        .doOnCompleted(() -> {
                            log.info("cleanup jobDiscoveryInfo onCompleted for {}", lookupKey);
                            discoveryInfoBehaviorSubject.toSerialized().onCompleted();
                            doOnZeroConnections.call(lookupKey);
                        })
                        .subscribeOn(scheduler)
                        .subscribe((schedInfo) -> discoveryInfoBehaviorSubject.onNext(schedInfo));
                initComplete.countDown();

            } else {
                try {
                    initComplete.await();
                } catch (InterruptedException e) {
                    log.error(e.getMessage());
                }
            }
        }
/**
         * If invoked the first time it will subscribe to the schedulingInfo Observable via mantisClient and onNext
         * the results to the schedulinginfoSubject
         * If 2 or more threads concurrently invoke this only 1 will do the initialization while others wait.
         */
If invoked the first time it will subscribe to the schedulingInfo Observable via mantisClient and onNext
the results to the schedulinginfoSubject
If 2 or more threads concurrently invoke this only 1 will do the initialization while others wait.
If invoked the first time it will subscribe to the schedulingInfo Observable via mantisClient and onNext the results to the schedulinginfoSubject If 2 or more threads concurrently invoke this only 1 will do the initialization while others wait .
private void init() {
            if (!inited.getAndSet(true)) {
                Observable<JobSchedulingInfo> jobSchedulingInfoObs;
                switch (lookupKey.getLookupType()) {
                    case JOB_ID:
                        jobSchedulingInfoObs = mantisClient.getSchedulingChanges(lookupKey.getId());
                        break;
                    case JOB_CLUSTER:
                        jobSchedulingInfoObs = mantisClient.jobClusterDiscoveryInfoStream(lookupKey.getId());
                        break;
                    default:
                        throw new IllegalArgumentException("lookup key type is not supported " + lookupKey.getLookupType());
                }
                subscription = jobSchedulingInfoObs
                        .retryWhen(Util.getRetryFunc(log, "job scheduling info for (" + lookupKey.getLookupType() + ") " + lookupKey.id))
                        .doOnError((t) -> {
                            log.info("cleanup jobDiscoveryInfo onError for {}", lookupKey);
                            discoveryInfoBehaviorSubject.toSerialized().onError(t);
                            doOnZeroConnections.call(lookupKey);
                        })
                        .doOnCompleted(() -> {
                            log.info("cleanup jobDiscoveryInfo onCompleted for {}", lookupKey);
                            discoveryInfoBehaviorSubject.toSerialized().onCompleted();
                            doOnZeroConnections.call(lookupKey);
                        })
                        .subscribeOn(scheduler)
                        .subscribe((schedInfo) -> discoveryInfoBehaviorSubject.onNext(schedInfo));
                initComplete.countDown();
            } else {
                try {
                    initComplete.await();
                } catch (InterruptedException e) {
                    log.error(e.getMessage());
                }
            }
        }
private void init ( ) { if ( ! inited . getAndSet ( true ) ) { Observable < JobSchedulingInfo > jobSchedulingInfoObs ; switch ( lookupKey . getLookupType ( ) ) { case JOB_ID : jobSchedulingInfoObs = mantisClient . getSchedulingChanges ( lookupKey . getId ( ) ) ; break ; case JOB_CLUSTER : jobSchedulingInfoObs = mantisClient . jobClusterDiscoveryInfoStream ( lookupKey . getId ( ) ) ; break ; default : throw new IllegalArgumentException ( "lookup key type is not supported " + lookupKey . getLookupType ( ) ) ; } subscription = jobSchedulingInfoObs . retryWhen ( Util . getRetryFunc ( log , "job scheduling info for (" + lookupKey . getLookupType ( ) + ") " + lookupKey . id ) ) . doOnError ( ( t ) -> { log . info ( "cleanup jobDiscoveryInfo onError for {}" , lookupKey ) ; discoveryInfoBehaviorSubject . toSerialized ( ) . onError ( t ) ; doOnZeroConnections . call ( lookupKey ) ; } ) . doOnCompleted ( ( ) -> { log . info ( "cleanup jobDiscoveryInfo onCompleted for {}" , lookupKey ) ; discoveryInfoBehaviorSubject . toSerialized ( ) . onCompleted ( ) ; doOnZeroConnections . call ( lookupKey ) ; } ) . subscribeOn ( scheduler ) . subscribe ( ( schedInfo ) -> discoveryInfoBehaviorSubject . onNext ( schedInfo ) ) ; initComplete . countDown ( ) ; } else { try { initComplete . await ( ) ; } catch ( InterruptedException e ) { log . error ( e . getMessage ( ) ) ; } } }
If invoked the first time it will subscribe to the schedulingInfo Observable via mantisClient and onNext
 the results to the schedulinginfoSubject
 If 2 or more threads concurrently invoke this only 1 will do the initialization while others wait.
If invoked the first time it will subscribe to the schedulingInfo Observable via mantisClient and onNext the results to the schedulinginfoSubject If 2 or more threads concurrently invoke this only 1 will do the initialization while others wait .