@Test public void testOpenBucketTimeout() { long delay = bucket.getBucketOpenTimeoutMillis() * 3; when(mockClusterManager.hasBucket(anyString())).thenReturn(Observable.just(true)); when(mockCluster.openBucket(anyString())).thenReturn(Observable.just(mockBucket) .delay(delay, TimeUnit.MILLISECONDS)); AsyncBucketHystrixObservableCommand cmd = new AsyncBucketHystrixObservableCommand(mockClusterFactory, couchbaseProperties); cmd.queryExecutor = mockQueryExecutor; cmd.scheduler = testScheduler; TestSubscriber<AsyncBucket> subscriber = new TestSubscriber<>(); cmd.observe() .subscribe(subscriber); testScheduler.advanceTimeBy(delay, TimeUnit.MILLISECONDS); subscriber.assertNoValues(); subscriber.assertNotCompleted(); }
private final Observable<AsyncBucket> openBucket(AsyncCluster cluster) { CouchbaseProperties.BucketProperties bucket = couchbaseProperties.getBucket(); log.info("Opening bucket: {}.", bucket.getName()); return cluster.openBucket(bucket.getName()) .doOnCompleted(() -> log.info("Successfully opened bucket: {}.", bucket.getName())) .doOnError(t -> log.error("Failed to open bucket: {}.", bucket.getName(), t)) .timeout(bucket.getBucketOpenTimeoutMillis(), MILLISECONDS); }
AsyncBucketHystrixObservableCommand extends HystrixObservableCommand<AsyncBucket> { private final Observable<AsyncBucket> openBucket(AsyncCluster cluster) { CouchbaseProperties.BucketProperties bucket = couchbaseProperties.getBucket(); log.info("Opening bucket: {}.", bucket.getName()); return cluster.openBucket(bucket.getName()) .doOnCompleted(() -> log.info("Successfully opened bucket: {}.", bucket.getName())) .doOnError(t -> log.error("Failed to open bucket: {}.", bucket.getName(), t)) .timeout(bucket.getBucketOpenTimeoutMillis(), MILLISECONDS); } }
AsyncBucketHystrixObservableCommand extends HystrixObservableCommand<AsyncBucket> { private final Observable<AsyncBucket> openBucket(AsyncCluster cluster) { CouchbaseProperties.BucketProperties bucket = couchbaseProperties.getBucket(); log.info("Opening bucket: {}.", bucket.getName()); return cluster.openBucket(bucket.getName()) .doOnCompleted(() -> log.info("Successfully opened bucket: {}.", bucket.getName())) .doOnError(t -> log.error("Failed to open bucket: {}.", bucket.getName(), t)) .timeout(bucket.getBucketOpenTimeoutMillis(), MILLISECONDS); } AsyncBucketHystrixObservableCommand(
            CouchbaseAsyncClusterFactory couchbaseAsyncClusterFactory,
            CouchbaseProperties couchbaseProperties,
            Setter setter
    );  AsyncBucketHystrixObservableCommand(
            CouchbaseAsyncClusterFactory couchbaseAsyncClusterFactory,
            CouchbaseProperties couchbaseProperties
    ); }
AsyncBucketHystrixObservableCommand extends HystrixObservableCommand<AsyncBucket> { private final Observable<AsyncBucket> openBucket(AsyncCluster cluster) { CouchbaseProperties.BucketProperties bucket = couchbaseProperties.getBucket(); log.info("Opening bucket: {}.", bucket.getName()); return cluster.openBucket(bucket.getName()) .doOnCompleted(() -> log.info("Successfully opened bucket: {}.", bucket.getName())) .doOnError(t -> log.error("Failed to open bucket: {}.", bucket.getName(), t)) .timeout(bucket.getBucketOpenTimeoutMillis(), MILLISECONDS); } AsyncBucketHystrixObservableCommand(
            CouchbaseAsyncClusterFactory couchbaseAsyncClusterFactory,
            CouchbaseProperties couchbaseProperties,
            Setter setter
    );  AsyncBucketHystrixObservableCommand(
            CouchbaseAsyncClusterFactory couchbaseAsyncClusterFactory,
            CouchbaseProperties couchbaseProperties
    );  }
AsyncBucketHystrixObservableCommand extends HystrixObservableCommand<AsyncBucket> { private final Observable<AsyncBucket> openBucket(AsyncCluster cluster) { CouchbaseProperties.BucketProperties bucket = couchbaseProperties.getBucket(); log.info("Opening bucket: {}.", bucket.getName()); return cluster.openBucket(bucket.getName()) .doOnCompleted(() -> log.info("Successfully opened bucket: {}.", bucket.getName())) .doOnError(t -> log.error("Failed to open bucket: {}.", bucket.getName(), t)) .timeout(bucket.getBucketOpenTimeoutMillis(), MILLISECONDS); } AsyncBucketHystrixObservableCommand(
            CouchbaseAsyncClusterFactory couchbaseAsyncClusterFactory,
            CouchbaseProperties couchbaseProperties,
            Setter setter
    );  AsyncBucketHystrixObservableCommand(
            CouchbaseAsyncClusterFactory couchbaseAsyncClusterFactory,
            CouchbaseProperties couchbaseProperties
    );   }