CouchbaseTableWriteFunction extends BaseCouchbaseTableFunction<V> implements TableWriteFunction<String, V> { @Override public CompletableFuture<Void> putAsync(String key, V record) { Preconditions.checkArgument(StringUtils.isNotBlank(key), "key must not be null, empty or blank"); Preconditions.checkArgument(!key.contains(" "), String.format("key should not contain spaces: %s", key)); Preconditions.checkNotNull(record); Document<?> document = record instanceof JsonObject ? JsonDocument.create(key, (int) ttl.getSeconds(), (JsonObject) record) : BinaryDocument.create(key, (int) ttl.getSeconds(), Unpooled.copiedBuffer(valueSerde.toBytes(record))); return asyncWriteHelper( bucket.async().upsert(document, timeout.toMillis(), TimeUnit.MILLISECONDS), String.format("Failed to insert key %s into bucket %s", key, bucketName)); } CouchbaseTableWriteFunction(String bucketName, Class<V> valueClass, String... clusterNodes); @Override void init(Context context, AsyncReadWriteTable table); @Override CompletableFuture<Void> putAsync(String key, V record); @Override CompletableFuture<Void> deleteAsync(String key);  }
@Test(expected = IllegalArgumentException.class) public void testPutAsyncNullKey() { CouchbaseTableWriteFunction<JsonObject> writeFunction = createAndInit(); writeFunction.putAsync(null, JsonObject.create()); }
@Test(expected = IllegalArgumentException.class) public void testPutAsyncKeyWithSpaces() { CouchbaseTableWriteFunction<JsonObject> writeFunction = createAndInit(); writeFunction.putAsync("k e y", JsonObject.create()); }
@Test(expected = NullPointerException.class) public void testPutAsyncNullValue() { CouchbaseTableWriteFunction<JsonObject> writeFunction = createAndInit(); writeFunction.putAsync("key", null); }
@Test public void testPutAsyncException() { String key = "throwExceptionKey"; JsonObject value = JsonObject.create(); Bucket bucket = mock(Bucket.class); AsyncBucket asyncBucket = mock(AsyncBucket.class); CouchbaseTableWriteFunction<JsonObject> writeFunction = createAndInit(bucket, asyncBucket); when(asyncBucket.upsert(any(Document.class), anyLong(), any(TimeUnit.class))).thenReturn( Observable.error(new CouchbaseException())); assertTrue(writeFunction.putAsync(key, value).isCompletedExceptionally()); }
@Test public void testPutAsyncJsonObjectValue() throws Exception { String key = "key"; JsonObject value = JsonObject.fromJson("{\"id\": 1}"); Bucket bucket = mock(Bucket.class); AsyncBucket asyncBucket = mock(AsyncBucket.class); CouchbaseTableWriteFunction<JsonObject> writeFunction = createAndInit(bucket, asyncBucket); when(asyncBucket.upsert(any(Document.class), anyLong(), any(TimeUnit.class))).thenReturn(Observable.just(null)); assertNull(writeFunction.putAsync(key, value).get()); }
@Test public void testPutAsyncStringValue() throws Exception { String key = "key"; String value = "value"; Bucket bucket = mock(Bucket.class); AsyncBucket asyncBucket = mock(AsyncBucket.class); CouchbaseTableWriteFunction<String> writeFunction = createAndInit(String.class, new StringSerde(), bucket, asyncBucket); when(asyncBucket.upsert(any(Document.class), anyLong(), any(TimeUnit.class))).thenReturn(Observable.just(null)); assertNull(writeFunction.putAsync(key, value).get()); }