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 ; } } ) ; }
utility methods for mutations