Insert

The insert method allows you to store a Document, if it does not already exist in the bucket. If it does exist, the Observable fails with a DocumentAlreadyExistsException (or it is thrown in the synchronous counterpart).

JsonObject content = JsonObject.empty().put("name", "Michael");
JsonDocument doc = JsonDocument.create("docId", content);
JsonDocument inserted = bucket.insert(doc)

JsonObject content = JsonObject.empty().put("name", "Michael");
JsonDocument doc = JsonDocument.create("docId", content);
Observable<JsonDocument> inserted = asyncBucket.insert(doc)

If the Document also has the expiry time set, it will be respected and picked up by the server.

It doesn't matter what type of Document is inserted, it's type is inferred from the method argument and the corresponding Transcoder is used to encode it.

The Document returned as a result is a different one compare to the Document passed in. It references some values like its id and content, but also has the CAS value set.

Upsert

The upsert method works similar to insert, but it also overrides a already stored Document (so there is no DocumentAlreadyExistsException thrown.

JsonObject content = JsonObject.empty().put("name", "Michael");
JsonDocument doc = JsonDocument.create("docId", content);
JsonDocument inserted = bucket.upsert(doc)

JsonObject content = JsonObject.empty().put("name", "Michael");
JsonDocument doc = JsonDocument.create("docId", content);
Observable<JsonDocument> inserted = asyncBucket.upsert(doc)

If the Document also has the expiry time set, it will be respected and picked up by the server.

It doesn't matter what type of Document is upserted, it's type is inferred from the method argument and the corresponding Transcoder is used to encode it.

The Document returned as a result is a different one compare to the Document passed in. It references some values like its id and content, but also has the CAS value set.

Durability Requirements

If no durability requirements are set on the insert or upsert methods, the operation will succeed when the server acknowledges the document in its managed cache layer. While this is a performant operation, there might be situations where you want to make sure that your document has been persisted and/or replicated so that it survives power outages and other node failures.

Both methods provide overloads to supply such requirements:

Observable<D> insert(D document, PersistTo persistTo);
Observable<D> insert(D document, ReplicateTo replicateTo);
Observable<D> insert(D document, PersistTo persistTo, ReplicateTo replicateTo);

Observable<D> upsert(D document, PersistTo persistTo);
Observable<D> upsert(D document, ReplicateTo replicateTo);
Observable<D> upsert(D document, PersistTo persistTo, ReplicateTo replicateTo);

In addition, the synchronous wrapper provides the same methods with custom timeouts.

You can configure either just one or both of the requirements when inserting or upserting. From an application point of view nothing needs to be changed when working with the response, although there is something that need to be kept in mind:

The internal implementation first performs a regular insert or upsert operation and afterwards starts polling the specific affected cluster nodes for the state of the document. If something fails during this operation (and failing the Observable), the original operation might have succeeded nonetheless.

// Insert the document and make sure it is persisted to the master node
bucket.insert(document, PersistTo.MASTER);

// Insert the document and make sure it is replicate to one replica node
bucket.insert(document, ReplicateTo.ONE);

// Insert the documen and make sure it is persisted to one node and replicated to two
bucket.insert(document, PersistTo.ONE, ReplicateTo.TWO);

Batching

Because everything is asynchronous by default, batching inserts or upserts can be achieved with Observable functionality.

A combination of just() and flatMap() is used to store them without blocking:

JsonDocument doc1 = JsonDocument.create("id1", content);
JsonDocument doc2 = JsonDocument.create("id2", content);
JsonDocument doc3 = JsonDocument.create("id3", content);

Observable
    .just(doc1, doc2, doc3)
    .flatMap(new Func1<JsonDocument, Observable<JsonDocument>>() {
        @Override
        public Observable<JsonDocument> call(JsonDocument document) {
            return bucket.insert(document);
        }
    }).subscribe();

For the blocking API, batching is currently not supported. It is recommended to fall back on the asynchronous API for now. We are considering adding a bulk API soon.