import com.couchbase.client.java.document.Document; //导入依赖的package包/类
@Override
public Tt upsert(Tt tt) {
    JsonObject content = JsonObject.fromJson(gson.toJson(tt));
    content.put(DOCTYPE_KEY, DOCTYPE_TT);
    Document<?> document = JsonDocument.create(tt.getId().toString(), content);
    bucket.upsert(document);
    return tt;
}
 

import com.couchbase.client.java.document.Document; //导入依赖的package包/类
private <T> Document<T> toDocument(final Object obj) {
    final Class<?> cls = obj.getClass();

    if (Document.class.isAssignableFrom(cls)) {
        return (Document<T>) obj;
    } else {
        return (Document<T>) toJsonDocument(obj);
    }
}
 

import com.couchbase.client.java.document.Document; //导入依赖的package包/类
private <T> T toEntityForUpdate(Class<T> cls, final Document<?> document) {
    if (cls.isAssignableFrom(document.getClass())) {
        return (T) document;
    } else {
        return toEntity(cls, (JsonDocument) document);
    }
}
 

import com.couchbase.client.java.document.Document; //导入依赖的package包/类
@Override
public Observable<Document> startImport() {
	MongoClient client = MongoClients.create(connectionString);
	MongoDatabase db = client.getDatabase(dbName);
	return db.getCollection(collectionName).find().toObservable()
			.map(new Func1<org.bson.Document, Document>() {
				public Document call(org.bson.Document mongoDoc) {
					mongoDoc.put(typeField, type);
					RawJsonDocument d = RawJsonDocument.create(mongoDoc
							.getObjectId("_id").toHexString(), mongoDoc
							.toJson());
					return d;
				};
			});
}
 

import com.couchbase.client.java.document.Document; //导入依赖的package包/类
public Observable<Document> createNode(String[] line) {
	ObjectNode node = objectMapper.createObjectNode();
	for (int i = 0; i < getTotalColumn(); i++) {
		String columnType = getColumType().get(i);
		RowType.valueOf(columnType).addField(node, columName.get(i),
				line[i], getSimpleDateFormat());
	}
	String kc = getKeyColum();
	String key = getKeyPrefix() + node.get(kc).asText();
	RawJsonDocument rjd = RawJsonDocument.create(key, node.toString());
	return Observable.just(rjd);
}
 

import com.couchbase.client.java.document.Document; //导入依赖的package包/类
@Override
public Observable<? extends Document> call(Document doc) {
	return asyncBucket
			.upsert(doc)
			.timeout(importTimeout, TimeUnit.MILLISECONDS)
			.retryWhen(
					RetryBuilder
							.anyOf(RequestCancelledException.class)
							.delay(fixed(requestCancelledExceptionDelay,
									TimeUnit.MILLISECONDS))
							.max(requestCancelledExceptionRetries).build())
			.retryWhen(
					RetryBuilder
							.anyOf(TemporaryFailureException.class,
									BackpressureException.class)
							.delay(fixed(temporaryFailureExceptionDelay,
									TimeUnit.MILLISECONDS))
							.max(temporaryFailureExceptionRetries).build())
			.doOnError(t -> writeToErrorLog(doc.id()))
			.doOnNext(jd -> writeToSuccessLog(doc.id()))
			.onErrorResumeNext(
					new Func1<Throwable, Observable<Document>>() {
						@Override
						public Observable<Document> call(Throwable throwable) {
							log.error(String.format(
									"Could not import document ", doc.id()));
							log.error(throwable);
							return Observable.empty();
						}
					});
}
 

import com.couchbase.client.java.document.Document; //导入依赖的package包/类
@Override
public Observable<Document> startImport() {
	throw new UnsupportedOperationException();
}
 

import com.couchbase.client.java.document.Document; //导入依赖的package包/类
public CouchbaseWriter(CouchbaseEnvironment couchbaseEnvironment, Config config) {

    List<String> hosts = ConfigUtils.getStringList(config, CouchbaseWriterConfigurationKeys.BOOTSTRAP_SERVERS);

    _cluster = CouchbaseCluster.create(couchbaseEnvironment, hosts);

    String bucketName = ConfigUtils.getString(config, CouchbaseWriterConfigurationKeys.BUCKET,
        CouchbaseWriterConfigurationKeys.BUCKET_DEFAULT);

    String password = ConfigUtils.getString(config, CouchbaseWriterConfigurationKeys.PASSWORD, "");

    _bucket = _cluster.openBucket(bucketName, password,
        Collections.<Transcoder<? extends Document, ?>>singletonList(_tupleDocumentTranscoder));

    _operationTimeout = ConfigUtils.getLong(config, CouchbaseWriterConfigurationKeys.OPERATION_TIMEOUT_MILLIS,
        CouchbaseWriterConfigurationKeys.OPERATION_TIMEOUT_DEFAULT);
    _operationTimeunit = TimeUnit.MILLISECONDS;

    _defaultWriteResponseMapper = new GenericWriteResponseWrapper<>();

    log.info("Couchbase writer configured with: hosts: {}, bucketName: {}, operationTimeoutInMillis: {}",
        hosts, bucketName, _operationTimeout);
  }
 

import com.couchbase.client.java.document.Document; //导入依赖的package包/类
@Test
public void testUpdate() throws Exception {
    when(bucket.replace(any(Document.class))).thenReturn(any(Document.class));
    impl.update("ABC123", new Tester("testerName", "testerOtherProperty"));
    verify(bucket).replace(RawJsonDocument.create("TESTER:ABC123", fixture("fixtures/tester.json")));
}
 

import com.couchbase.client.java.document.Document; //导入依赖的package包/类
Observable<Document> startImport() throws Exception;