private Observable < AbstractDocument > writeDoc ( String key , int ttl , long cas , ByteArrayOutputStream baos , Record record ) { WriteOperationType opType = getOperationFromHeader ( record , key ) ; if ( opType == null ) { return Observable . empty ( ) ; } AbstractDocument doc ; if ( config . dataFormat == DataFormat . JSON ) { try { doc = JsonDocument . create ( key , ttl , JsonObject . fromJson ( baos . toString ( config . dataFormatConfig . charset ) ) , cas ) ; } catch ( Exception e ) { return handleError ( record , Errors . COUCHBASE_10 , e ) ; } } else { doc = ByteArrayDocument . create ( key , ttl , baos . toByteArray ( ) , cas ) ; } switch ( opType ) { case DELETE : { LOG . debug ( "DELETE key: {}, TTL: {}, CAS: {}" , key , ttl , cas ) ; return connector . bucket ( ) . remove ( doc , config . persistTo , config . replicateTo ) . timeout ( config . couchbase . kvTimeout , TimeUnit . MILLISECONDS ) ; } case INSERT : { LOG . debug ( "INSERT key: {}, TTL: {}, CAS: {}" , key , ttl , cas ) ; return connector . bucket ( ) . insert ( doc , config . persistTo , config . replicateTo ) . timeout ( config . couchbase . kvTimeout , TimeUnit . MILLISECONDS ) ; } case REPLACE : { LOG . debug ( "REPLACE key: {}, TTL: {}, CAS: {}" , key , ttl , cas ) ; return connector . bucket ( ) . replace ( doc , config . persistTo , config . replicateTo ) . timeout ( config . couchbase . kvTimeout , TimeUnit . MILLISECONDS ) ; } case UPSERT : { LOG . debug ( "UPSERT key: {}, TTL: {}, CAS: {}" , key , ttl , cas ) ; return connector . bucket ( ) . upsert ( doc , config . persistTo , config . replicateTo ) . timeout ( config . couchbase . kvTimeout , TimeUnit . MILLISECONDS ) ; } default : return Observable . empty ( ) ; } }
Executes a document write operation .