@Test public void testExecuteN1qlQueryError() { when(mockResult.finalSuccess()).thenReturn(Observable.just(false)); JsonObject jsonObject = JsonObject.fromJson("{\"test\" : \"error\"}"); when(mockResult.errors()).thenReturn(Observable.just(jsonObject)); TestSubscriber<AsyncN1qlQueryRow> subscriber = new TestSubscriber<>(); executeN1qlQuery(mockBucket, mockQuery) .subscribe(subscriber); subscriber.assertError(CouchbaseException.class); verify(mockResult, never()).rows(); }
public static Observable<AsyncN1qlQueryRow> executeN1qlQuery(AsyncBucket bucket, N1qlQuery query) { log.debug("Executing query: {}.", query.statement()); return bucket.query(query) .map(result -> Tuple.create(query, result)) .flatMap(tuple1 -> { AsyncN1qlQueryResult result = tuple1.value2(); Statement statement = tuple1.value1().statement(); return result .finalSuccess() .defaultIfEmpty(false) .zipWith(result.info(), Tuple::create) .flatMap(tuple2 -> { boolean success = tuple2.value1(); N1qlMetrics metrics = tuple2.value2(); log.debug("Query metrics: {}.", metrics); if (success) { log.debug("Successfully executed query: {}.", statement); return result.rows(); } else { return result.errors() .map(error -> { String errorMessage = String.format("Failed to execute query: %s, " + "error: %s.", statement, error.toString()); return new CouchbaseException(errorMessage); }) .flatMap(Observable::error); } }); }); }
CouchbaseQueryUtil { public static Observable<AsyncN1qlQueryRow> executeN1qlQuery(AsyncBucket bucket, N1qlQuery query) { log.debug("Executing query: {}.", query.statement()); return bucket.query(query) .map(result -> Tuple.create(query, result)) .flatMap(tuple1 -> { AsyncN1qlQueryResult result = tuple1.value2(); Statement statement = tuple1.value1().statement(); return result .finalSuccess() .defaultIfEmpty(false) .zipWith(result.info(), Tuple::create) .flatMap(tuple2 -> { boolean success = tuple2.value1(); N1qlMetrics metrics = tuple2.value2(); log.debug("Query metrics: {}.", metrics); if (success) { log.debug("Successfully executed query: {}.", statement); return result.rows(); } else { return result.errors() .map(error -> { String errorMessage = String.format("Failed to execute query: %s, " + "error: %s.", statement, error.toString()); return new CouchbaseException(errorMessage); }) .flatMap(Observable::error); } }); }); } }
CouchbaseQueryUtil { public static Observable<AsyncN1qlQueryRow> executeN1qlQuery(AsyncBucket bucket, N1qlQuery query) { log.debug("Executing query: {}.", query.statement()); return bucket.query(query) .map(result -> Tuple.create(query, result)) .flatMap(tuple1 -> { AsyncN1qlQueryResult result = tuple1.value2(); Statement statement = tuple1.value1().statement(); return result .finalSuccess() .defaultIfEmpty(false) .zipWith(result.info(), Tuple::create) .flatMap(tuple2 -> { boolean success = tuple2.value1(); N1qlMetrics metrics = tuple2.value2(); log.debug("Query metrics: {}.", metrics); if (success) { log.debug("Successfully executed query: {}.", statement); return result.rows(); } else { return result.errors() .map(error -> { String errorMessage = String.format("Failed to execute query: %s, " + "error: %s.", statement, error.toString()); return new CouchbaseException(errorMessage); }) .flatMap(Observable::error); } }); }); } private  CouchbaseQueryUtil(); }
CouchbaseQueryUtil { public static Observable<AsyncN1qlQueryRow> executeN1qlQuery(AsyncBucket bucket, N1qlQuery query) { log.debug("Executing query: {}.", query.statement()); return bucket.query(query) .map(result -> Tuple.create(query, result)) .flatMap(tuple1 -> { AsyncN1qlQueryResult result = tuple1.value2(); Statement statement = tuple1.value1().statement(); return result .finalSuccess() .defaultIfEmpty(false) .zipWith(result.info(), Tuple::create) .flatMap(tuple2 -> { boolean success = tuple2.value1(); N1qlMetrics metrics = tuple2.value2(); log.debug("Query metrics: {}.", metrics); if (success) { log.debug("Successfully executed query: {}.", statement); return result.rows(); } else { return result.errors() .map(error -> { String errorMessage = String.format("Failed to execute query: %s, " + "error: %s.", statement, error.toString()); return new CouchbaseException(errorMessage); }) .flatMap(Observable::error); } }); }); } private  CouchbaseQueryUtil(); static Observable<AsyncN1qlQueryRow> executeN1qlQuery(AsyncBucket bucket, N1qlQuery query); }
CouchbaseQueryUtil { public static Observable<AsyncN1qlQueryRow> executeN1qlQuery(AsyncBucket bucket, N1qlQuery query) { log.debug("Executing query: {}.", query.statement()); return bucket.query(query) .map(result -> Tuple.create(query, result)) .flatMap(tuple1 -> { AsyncN1qlQueryResult result = tuple1.value2(); Statement statement = tuple1.value1().statement(); return result .finalSuccess() .defaultIfEmpty(false) .zipWith(result.info(), Tuple::create) .flatMap(tuple2 -> { boolean success = tuple2.value1(); N1qlMetrics metrics = tuple2.value2(); log.debug("Query metrics: {}.", metrics); if (success) { log.debug("Successfully executed query: {}.", statement); return result.rows(); } else { return result.errors() .map(error -> { String errorMessage = String.format("Failed to execute query: %s, " + "error: %s.", statement, error.toString()); return new CouchbaseException(errorMessage); }) .flatMap(Observable::error); } }); }); } private  CouchbaseQueryUtil(); static Observable<AsyncN1qlQueryRow> executeN1qlQuery(AsyncBucket bucket, N1qlQuery query);  }