// writeMessage writes one message to the destination mongo, or sends an error down the pipe
// TODO this can be cleaned up.  I'm not sure whether this should pipe the error, or whether the
//   caller should pipe the error
writeMessage writes one message to the destination mongo, or sends an error down the pipe
TODO this can be cleaned up.  I'm not sure whether this should pipe the error, or whether the
caller should pipe the error
// writeMessage writes one message to the destination mongo or sends an error down the pipe // TODO this can be cleaned up . I ' m not sure whether this should pipe the error or whether the // caller should pipe the error
func (m *Mongodb) writeMessage(msg *message.Msg) (*message.Msg, error) {
	_, msgColl, err := msg.SplitNamespace()
	if err != nil {
		m.pipe.Err <- adaptor.NewError(adaptor.ERROR, m.path, fmt.Sprintf("mongodb error (msg namespace improperly formatted, must be database.collection, got %s)", msg.Namespace), msg.Data)
		return msg, nil
	}
	collection := m.mongoSession.DB(m.database).C(msgColl)
	if !msg.IsMap() {
		m.pipe.Err <- adaptor.NewError(adaptor.ERROR, m.path, fmt.Sprintf("mongodb error (document must be a bson document, got %T instead)", msg.Data), msg.Data)
		return msg, nil
	}
	doc := &syncDoc{
		Doc:        msg.Map(),
		Collection: msgColl,
	}
	if m.bulk {
		m.bulkWriteChannel <- doc
	} else if msg.Op == message.Delete {
		err := collection.Remove(doc.Doc)
		if err != nil {
			m.pipe.Err <- adaptor.NewError(adaptor.ERROR, m.path, fmt.Sprintf("mongodb error removing (%s)", err.Error()), msg.Data)
		}
	} else {
		err := collection.Insert(doc.Doc)
		if mgo.IsDup(err) {
			err = collection.Update(bson.M{"_id": doc.Doc["_id"]}, doc.Doc)
		}
		if err != nil {
			m.pipe.Err <- adaptor.NewError(adaptor.ERROR, m.path, fmt.Sprintf("mongodb error (%s)", err.Error()), msg.Data)
		}
	}
	return msg, nil
}
func ( m * Mongodb ) writeMessage ( msg * message . Msg ) ( * message . Msg , error ) { _ , msgColl , err := msg . SplitNamespace ( ) if err != nil { m . pipe . Err <- adaptor . NewError ( adaptor . ERROR , m . path , fmt . Sprintf ( " " , msg . Namespace ) , msg . Data ) return msg , nil } collection := m . mongoSession . DB ( m . database ) . C ( msgColl ) if ! msg . IsMap ( ) { m . pipe . Err <- adaptor . NewError ( adaptor . ERROR , m . path , fmt . Sprintf ( " " , msg . Data ) , msg . Data ) return msg , nil } doc := & syncDoc { Doc : msg . Map ( ) , Collection : msgColl , } if m . bulk { m . bulkWriteChannel <- doc } else if msg . Op == message . Delete { err := collection . Remove ( doc . Doc ) if err != nil { m . pipe . Err <- adaptor . NewError ( adaptor . ERROR , m . path , fmt . Sprintf ( " " , err . Error ( ) ) , msg . Data ) } } else { err := collection . Insert ( doc . Doc ) if mgo . IsDup ( err ) { err = collection . Update ( bson . M { " " : doc . Doc [ " " ] } , doc . Doc ) } if err != nil { m . pipe . Err <- adaptor . NewError ( adaptor . ERROR , m . path , fmt . Sprintf ( " " , err . Error ( ) ) , msg . Data ) } } return msg , nil }
writeMessage writes one message to the destination mongo, or sends an error down the pipe
TODO this can be cleaned up.
writeMessage writes one message to the destination mongo or sends an error down the pipe TODO this can be cleaned up .