code: private <T> Observable<DocumentFragment<T>> subdocObserveMutation(Observable<DocumentFragment<T>> mutation, final long timeout, final TimeUnit timeUnit) {
          if (persistTo == PersistTo.NONE && replicateTo == ReplicateTo.NONE) {
              return mutation;
          }
  
          return mutation.flatMap(new Func1<DocumentFragment<T>, Observable<DocumentFragment<T>>>() {
              @Override
              public Observable<DocumentFragment<T>> call(final DocumentFragment<T> frag) {
                  Observable<DocumentFragment<T>> result = Observe
                      .call(core, bucketName, frag.id(), frag.cas(), false, frag.mutationToken(), persistTo.value(), replicateTo.value(),
                          environment.observeIntervalDelay(), environment.retryStrategy())
                      .map(new Func1<Boolean, DocumentFragment<T>>() {
                          @Override
                          public DocumentFragment<T> call(Boolean aBoolean) {
                              return frag;
                          }
                      })
                      .onErrorResumeNext(new Func1<Throwable, Observable<DocumentFragment<T>>>() {
                          @Override
                          public Observable<DocumentFragment<T>> call(Throwable throwable) {
                              return Observable.error(new DurabilityException(
                                  "Durability requirement failed: " + throwable.getMessage(),
                                  throwable));
                          }
                      });
                  if (timeout > 0) {
                      result = result.timeout(timeout, timeUnit, environment.scheduler());
                  }
                  return result;
              }
          });
      }
docs: utility methods for mutations