CouchbaseTableReadFunction extends BaseCouchbaseTableFunction<V> implements TableReadFunction<String, V> { @Override public CompletableFuture<V> getAsync(String key) { Preconditions.checkArgument(StringUtils.isNotBlank(key), "key must not be null, empty or blank"); CompletableFuture<V> future = new CompletableFuture<>(); Single<? extends Document<?>> singleObservable = bucket.async().get(key, documentType, timeout.toMillis(), TimeUnit.MILLISECONDS).toSingle(); singleObservable.subscribe(new SingleSubscriber<Document<?>>() { @Override public void onSuccess(Document<?> document) { if (document != null) { if (document instanceof BinaryDocument) { handleGetAsyncBinaryDocument((BinaryDocument) document, future, key); } else { future.complete((V) document.content()); } } else { future.completeExceptionally(new SamzaException(String.format("Got unexpected null value from key %s", key))); } } @Override public void onError(Throwable throwable) { if (throwable instanceof NoSuchElementException) { future.complete(null); } else { future.completeExceptionally(new SamzaException(String.format("Failed to get key %s", key), throwable)); } } }); return future; } CouchbaseTableReadFunction(String bucketName, Class<V> valueClass, String... clusterNodes); @Override void init(Context context, AsyncReadWriteTable table); @Override CompletableFuture<V> getAsync(String key);  }
@Test(expected = IllegalArgumentException.class) public void testGetAsyncNullKey() { CouchbaseTableReadFunction readFunction = createAndInit(); readFunction.getAsync(null); }
@Test public void testGetAsyncException() { String key = "throwExceptionKey"; Bucket bucket = mock(Bucket.class); AsyncBucket asyncBucket = mock(AsyncBucket.class); CouchbaseTableReadFunction readFunction = createAndInit(bucket, asyncBucket); when(asyncBucket.get(eq(key), anyObject(), anyLong(), any(TimeUnit.class))).thenReturn( Observable.error(new CouchbaseException())); assertTrue(readFunction.getAsync(key).isCompletedExceptionally()); }
@Test public void testGetAsyncFailedToDeserialize() { String key = "key"; Bucket bucket = mock(Bucket.class); AsyncBucket asyncBucket = mock(AsyncBucket.class); CouchbaseTableReadFunction readFunction = createAndInit(String.class, new StringSerde(), bucket, asyncBucket); when(asyncBucket.get(eq(key), anyObject(), anyLong(), any(TimeUnit.class))).thenReturn( Observable.just(BinaryDocument.create(key, null))); assertTrue(readFunction.getAsync(key).isCompletedExceptionally()); }
@Test public void testGetAsyncNullValue() throws Exception { String key = "NonExistingKey"; Bucket bucket = mock(Bucket.class); AsyncBucket asyncBucket = mock(AsyncBucket.class); CouchbaseTableReadFunction readFunction = createAndInit(String.class, new StringSerde(), bucket, asyncBucket); when(asyncBucket.get(eq(key), anyObject(), anyLong(), any(TimeUnit.class))).thenReturn(Observable.empty()); assertNull(readFunction.getAsync(key).get()); }
@Test public void testGetAsyncNonExistingKey() throws Exception { String key = "NonExistingKey"; Bucket bucket = mock(Bucket.class); AsyncBucket asyncBucket = mock(AsyncBucket.class); CouchbaseTableReadFunction readFunction = createAndInit(String.class, new StringSerde(), bucket, asyncBucket); when(asyncBucket.get(eq(key), anyObject(), anyLong(), any(TimeUnit.class))).thenReturn(Observable.empty()); assertNull(readFunction.getAsync(key).get()); }
@Test public void testGetAsyncStringValue() throws Exception { String key = "key"; String value = "value"; StringSerde stringSerde = new StringSerde(); Bucket bucket = mock(Bucket.class); AsyncBucket asyncBucket = mock(AsyncBucket.class); CouchbaseTableReadFunction readFunction = createAndInit(String.class, stringSerde, bucket, asyncBucket); when(asyncBucket.get(eq(key), anyObject(), anyLong(), any(TimeUnit.class))).thenReturn( Observable.just(BinaryDocument.create(key, Unpooled.wrappedBuffer(stringSerde.toBytes(value))))); assertEquals(value, readFunction.getAsync(key).get()); }
@Test public void testGetAsyncJsonObjectValue() throws Exception { String key = "key"; JsonObject value = JsonObject.fromJson("{\"id\": 1}"); Bucket bucket = mock(Bucket.class); AsyncBucket asyncBucket = mock(AsyncBucket.class); CouchbaseTableReadFunction readFunction = createAndInit(bucket, asyncBucket); when(asyncBucket.get(eq(key), anyObject(), anyLong(), any(TimeUnit.class))).thenReturn( Observable.just(JsonDocument.create(key, value))); assertEquals(value, readFunction.getAsync(key).get()); }