@Test public void testMultiTupleDocumentWrite() throws IOException, DataConversionException, ExecutionException, InterruptedException { CouchbaseWriter writer = new CouchbaseWriter(_couchbaseEnvironment, ConfigFactory.empty()); try { final Schema dataRecordSchema = SchemaBuilder.record("Data").fields().name("data").type().bytesType().noDefault().name("flags").type().intType() .noDefault().endRecord(); final Schema schema = SchemaBuilder.record("TestRecord").fields().name("key").type().stringType().noDefault().name("data") .type(dataRecordSchema).noDefault().endRecord(); final int numRecords = 1000; int outstandingRequests = 99; Iterator<GenericRecord> recordIterator = new Iterator<GenericRecord>() { private int currentIndex; @Override public void remove() { } @Override public boolean hasNext() { return (currentIndex < numRecords); } @Override public GenericRecord next() { GenericData.Record testRecord = new GenericData.Record(schema); String testContent = "hello world" + currentIndex; GenericData.Record dataRecord = new GenericData.Record(dataRecordSchema); dataRecord.put("data", ByteBuffer.wrap(testContent.getBytes(Charset.forName("UTF-8")))); dataRecord.put("flags", 0); testRecord.put("key", "hello" + currentIndex); testRecord.put("data", dataRecord); currentIndex++; return testRecord; } }; long kvTimeout = 10000; TimeUnit kvTimeoutUnit = TimeUnit.MILLISECONDS; writeRecords(new TupleDocumentIterator(recordIterator), writer, outstandingRequests, kvTimeout, kvTimeoutUnit); } finally { writer.close(); } }
@Override public void close() { if (!_bucket.isClosed()) { try { _bucket.close(); } catch (Exception e) { log.warn("Failed to close bucket", e); } } try { _cluster.disconnect(); } catch (Exception e) { log.warn("Failed to disconnect from cluster", e); } }
CouchbaseWriter implements AsyncDataWriter<D>, SyncDataWriter<D> { @Override public void close() { if (!_bucket.isClosed()) { try { _bucket.close(); } catch (Exception e) { log.warn("Failed to close bucket", e); } } try { _cluster.disconnect(); } catch (Exception e) { log.warn("Failed to disconnect from cluster", e); } } }
CouchbaseWriter implements AsyncDataWriter<D>, SyncDataWriter<D> { @Override public void close() { if (!_bucket.isClosed()) { try { _bucket.close(); } catch (Exception e) { log.warn("Failed to close bucket", e); } } try { _cluster.disconnect(); } catch (Exception e) { log.warn("Failed to disconnect from cluster", e); } } CouchbaseWriter(CouchbaseEnvironment couchbaseEnvironment, Config config); }
CouchbaseWriter implements AsyncDataWriter<D>, SyncDataWriter<D> { @Override public void close() { if (!_bucket.isClosed()) { try { _bucket.close(); } catch (Exception e) { log.warn("Failed to close bucket", e); } } try { _cluster.disconnect(); } catch (Exception e) { log.warn("Failed to disconnect from cluster", e); } } CouchbaseWriter(CouchbaseEnvironment couchbaseEnvironment, Config config); @Override Future<WriteResponse> write(final D record, final WriteCallback callback); @Override void flush(); @Override void cleanup(); @Override WriteResponse write(D record); @Override void close(); }
CouchbaseWriter implements AsyncDataWriter<D>, SyncDataWriter<D> { @Override public void close() { if (!_bucket.isClosed()) { try { _bucket.close(); } catch (Exception e) { log.warn("Failed to close bucket", e); } } try { _cluster.disconnect(); } catch (Exception e) { log.warn("Failed to disconnect from cluster", e); } } CouchbaseWriter(CouchbaseEnvironment couchbaseEnvironment, Config config); @Override Future<WriteResponse> write(final D record, final WriteCallback callback); @Override void flush(); @Override void cleanup(); @Override WriteResponse write(D record); @Override void close();  }