@Test public void testCreateBucketIfAbsent() { when(mockClusterManager.hasBucket(anyString())).thenReturn(Observable.just(false)); when(mockClusterManager.insertBucket(any(BucketSettings.class))) .thenReturn(Observable.just(mock(BucketSettings.class))); AsyncBucketHystrixObservableCommand cmd = new AsyncBucketHystrixObservableCommand(mockClusterFactory, couchbaseProperties); cmd.queryExecutor = mockQueryExecutor; cmd.scheduler = testScheduler; TestSubscriber<AsyncBucket> subscriber = new TestSubscriber<>(); cmd.observe() .subscribe(subscriber); testScheduler.advanceTimeBy(bucket.getBucketOpenTimeoutMillis(), TimeUnit.MILLISECONDS); subscriber.assertCompleted(); subscriber.assertNoErrors(); subscriber.assertValueCount(1); verify(mockClusterManager).insertBucket(argThat(settings -> settings.name().equals(bucket.getName()) && settings.enableFlush() == bucket.isEnableFlush() && settings.quota() == bucket.getDefaultQuotaMB() && settings.indexReplicas() == bucket.isIndexReplicas() && bucket.isCreateIfMissing() )); verify(mockCluster).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
    );   }