// applyOp applies one operation to the database
applyOp applies one operation to the database
// applyOp applies one operation to the database
func (r *Rethinkdb) applyOp(msg *message.Msg) (*message.Msg, error) {
	_, msgTable, err := msg.SplitNamespace()
	if err != nil {
		r.pipe.Err <- adaptor.NewError(adaptor.ERROR, r.path, fmt.Sprintf("rethinkdb error (msg namespace improperly formatted, must be database.table, got %s)", msg.Namespace), msg.Data)
		return msg, nil
	}
	if !msg.IsMap() {
		r.pipe.Err <- adaptor.NewError(adaptor.ERROR, r.path, "rethinkdb error (document must be a json document)", msg.Data)
		return msg, nil
	}
	doc := msg.Map()
	var resp gorethink.WriteResponse
	switch msg.Op {
	case message.Delete:
		var id string
		id, err = msg.IDString("id")
		if err != nil {
			r.pipe.Err <- adaptor.NewError(adaptor.ERROR, r.path, "rethinkdb error (cannot delete an object with a nil id)", msg.Data)
			return msg, nil
		}
		resp, err = gorethink.Table(msgTable).Get(id).Delete().RunWrite(r.client)
	case message.Insert:
		resp, err = gorethink.Table(msgTable).Insert(doc).RunWrite(r.client)
	case message.Update:
		resp, err = gorethink.Table(msgTable).Insert(doc, gorethink.InsertOpts{Conflict: "replace"}).RunWrite(r.client)
	}
	if err != nil {
		r.pipe.Err <- adaptor.NewError(adaptor.ERROR, r.path, "rethinkdb error (%s)", err)
		return msg, nil
	}
	err = r.handleResponse(&resp)
	if err != nil {
		r.pipe.Err <- adaptor.NewError(adaptor.ERROR, r.path, "rethinkdb error (%s)", err)
	}
	return msg, nil
}
func ( r * Rethinkdb ) applyOp ( msg * message . Msg ) ( * message . Msg , error ) { _ , msgTable , err := msg . SplitNamespace ( ) if err != nil { r . pipe . Err <- adaptor . NewError ( adaptor . ERROR , r . path , fmt . Sprintf ( " " , msg . Namespace ) , msg . Data ) return msg , nil } if ! msg . IsMap ( ) { r . pipe . Err <- adaptor . NewError ( adaptor . ERROR , r . path , " " , msg . Data ) return msg , nil } doc := msg . Map ( ) var resp gorethink . WriteResponse switch msg . Op { case message . Delete : var id string id , err = msg . IDString ( " " ) if err != nil { r . pipe . Err <- adaptor . NewError ( adaptor . ERROR , r . path , " " , msg . Data ) return msg , nil } resp , err = gorethink . Table ( msgTable ) . Get ( id ) . Delete ( ) . RunWrite ( r . client ) case message . Insert : resp , err = gorethink . Table ( msgTable ) . Insert ( doc ) . RunWrite ( r . client ) case message . Update : resp , err = gorethink . Table ( msgTable ) . Insert ( doc , gorethink . InsertOpts { Conflict : " " } ) . RunWrite ( r . client ) } if err != nil { r . pipe . Err <- adaptor . NewError ( adaptor . ERROR , r . path , " " , err ) return msg , nil } err = r . handleResponse ( & resp ) if err != nil { r . pipe . Err <- adaptor . NewError ( adaptor . ERROR , r . path , " " , err ) } return msg , nil }
applyOp applies one operation to the database
applyOp applies one operation to the database