@Test public void testCreatePrimaryIndex() { AsyncBucketHystrixObservableCommand cmd = new AsyncBucketHystrixObservableCommand(mockClusterFactory, couchbaseProperties); cmd.queryExecutor = mockQueryExecutor; cmd.scheduler = testScheduler; TestSubscriber<AsyncN1qlQueryRow> subscriber = new TestSubscriber<>(); cmd.createPrimaryIndex(mockBucket) .subscribe(subscriber); testScheduler.advanceTimeBy(bucket.getBucketOpenTimeoutMillis(), TimeUnit.MILLISECONDS); subscriber.assertCompleted(); subscriber.assertNoErrors(); subscriber.assertValueCount(1); String expectedQueryStatement = "CREATE PRIMARY INDEX ON `test`"; verify(mockQueryExecutor).apply(eq(mockBucket), argThat(query -> query.statement().toString().equals(expectedQueryStatement))); }
Observable<AsyncN1qlQueryRow> createPrimaryIndex(AsyncBucket bucket) { CouchbaseProperties.BucketProperties bucketProperties = couchbaseProperties.getBucket(); Statement statement = Index.createPrimaryIndex() .on(bucketProperties.getName()); SimpleN1qlQuery query = N1qlQuery.simple(statement); return Observable.just(0L) .observeOn(scheduler) .doOnNext(i -> log.info("Creating primary index.")) .flatMap(i -> Observable.timer(bucketProperties.getBucketOpenTimeoutMillis(), MILLISECONDS, scheduler) .zipWith(queryExecutor.apply(bucket, query), (j, row) -> row)) .doOnCompleted(() -> log.info("Successfully created primary index.")) .doOnError(t -> log.error("Failed to create primary index.", t)); }
AsyncBucketHystrixObservableCommand extends HystrixObservableCommand<AsyncBucket> { Observable<AsyncN1qlQueryRow> createPrimaryIndex(AsyncBucket bucket) { CouchbaseProperties.BucketProperties bucketProperties = couchbaseProperties.getBucket(); Statement statement = Index.createPrimaryIndex() .on(bucketProperties.getName()); SimpleN1qlQuery query = N1qlQuery.simple(statement); return Observable.just(0L) .observeOn(scheduler) .doOnNext(i -> log.info("Creating primary index.")) .flatMap(i -> Observable.timer(bucketProperties.getBucketOpenTimeoutMillis(), MILLISECONDS, scheduler) .zipWith(queryExecutor.apply(bucket, query), (j, row) -> row)) .doOnCompleted(() -> log.info("Successfully created primary index.")) .doOnError(t -> log.error("Failed to create primary index.", t)); } }
AsyncBucketHystrixObservableCommand extends HystrixObservableCommand<AsyncBucket> { Observable<AsyncN1qlQueryRow> createPrimaryIndex(AsyncBucket bucket) { CouchbaseProperties.BucketProperties bucketProperties = couchbaseProperties.getBucket(); Statement statement = Index.createPrimaryIndex() .on(bucketProperties.getName()); SimpleN1qlQuery query = N1qlQuery.simple(statement); return Observable.just(0L) .observeOn(scheduler) .doOnNext(i -> log.info("Creating primary index.")) .flatMap(i -> Observable.timer(bucketProperties.getBucketOpenTimeoutMillis(), MILLISECONDS, scheduler) .zipWith(queryExecutor.apply(bucket, query), (j, row) -> row)) .doOnCompleted(() -> log.info("Successfully created primary index.")) .doOnError(t -> log.error("Failed to create primary index.", t)); } AsyncBucketHystrixObservableCommand(
            CouchbaseAsyncClusterFactory couchbaseAsyncClusterFactory,
            CouchbaseProperties couchbaseProperties,
            Setter setter
    );  AsyncBucketHystrixObservableCommand(
            CouchbaseAsyncClusterFactory couchbaseAsyncClusterFactory,
            CouchbaseProperties couchbaseProperties
    ); }
AsyncBucketHystrixObservableCommand extends HystrixObservableCommand<AsyncBucket> { Observable<AsyncN1qlQueryRow> createPrimaryIndex(AsyncBucket bucket) { CouchbaseProperties.BucketProperties bucketProperties = couchbaseProperties.getBucket(); Statement statement = Index.createPrimaryIndex() .on(bucketProperties.getName()); SimpleN1qlQuery query = N1qlQuery.simple(statement); return Observable.just(0L) .observeOn(scheduler) .doOnNext(i -> log.info("Creating primary index.")) .flatMap(i -> Observable.timer(bucketProperties.getBucketOpenTimeoutMillis(), MILLISECONDS, scheduler) .zipWith(queryExecutor.apply(bucket, query), (j, row) -> row)) .doOnCompleted(() -> log.info("Successfully created primary index.")) .doOnError(t -> log.error("Failed to create primary index.", t)); } AsyncBucketHystrixObservableCommand(
            CouchbaseAsyncClusterFactory couchbaseAsyncClusterFactory,
            CouchbaseProperties couchbaseProperties,
            Setter setter
    );  AsyncBucketHystrixObservableCommand(
            CouchbaseAsyncClusterFactory couchbaseAsyncClusterFactory,
            CouchbaseProperties couchbaseProperties
    );  }
AsyncBucketHystrixObservableCommand extends HystrixObservableCommand<AsyncBucket> { Observable<AsyncN1qlQueryRow> createPrimaryIndex(AsyncBucket bucket) { CouchbaseProperties.BucketProperties bucketProperties = couchbaseProperties.getBucket(); Statement statement = Index.createPrimaryIndex() .on(bucketProperties.getName()); SimpleN1qlQuery query = N1qlQuery.simple(statement); return Observable.just(0L) .observeOn(scheduler) .doOnNext(i -> log.info("Creating primary index.")) .flatMap(i -> Observable.timer(bucketProperties.getBucketOpenTimeoutMillis(), MILLISECONDS, scheduler) .zipWith(queryExecutor.apply(bucket, query), (j, row) -> row)) .doOnCompleted(() -> log.info("Successfully created primary index.")) .doOnError(t -> log.error("Failed to create primary index.", t)); } AsyncBucketHystrixObservableCommand(
            CouchbaseAsyncClusterFactory couchbaseAsyncClusterFactory,
            CouchbaseProperties couchbaseProperties,
            Setter setter
    );  AsyncBucketHystrixObservableCommand(
            CouchbaseAsyncClusterFactory couchbaseAsyncClusterFactory,
            CouchbaseProperties couchbaseProperties
    );   }