@Test public void testCreateBucketError() { when(mockClusterManager.hasBucket(anyString())).thenReturn(Observable.just(false)); AsyncBucketHystrixObservableCommand cmd = new AsyncBucketHystrixObservableCommand(mockClusterFactory, couchbaseProperties); cmd.queryExecutor = mockQueryExecutor; when(mockClusterManager.insertBucket(any(BucketSettings.class))) .thenThrow(new RuntimeException("test")); TestSubscriber<AsyncBucket> subscriber = new TestSubscriber<>(); cmd.observe() .subscribe(subscriber); subscriber.assertError(RuntimeException.class); subscriber.assertNotCompleted(); verify(mockCluster, never()).openBucket(bucket.getName()); }
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
    );   }