/*
 * tail the oplog
 */
tail the oplog
/ * * tail the oplog * /
func (m *Mongodb) tailData() (err error) {
	var (
		collection = m.mongoSession.DB("local").C("oplog.rs")
		result     oplogDoc 
		query      = bson.M{
			"ts": bson.M{"$gte": m.oplogTime},
		}
		iter = collection.Find(query).LogReplay().Sort("$natural").Tail(m.oplogTimeout)
	)
	for {
		for iter.Next(&result) {
			if stop := m.pipe.Stopped; stop {
				return
			}
			if result.validOp() {
				db, coll, _ := m.splitNamespace(result.Ns)
				if db != m.database {
					continue
				}
				if strings.HasPrefix(coll, "system.") {
					continue
				} else if match := m.collectionMatch.MatchString(coll); !match {
					continue
				}
				var doc bson.M
				switch result.Op {
				case "i":
					doc = result.O
				case "d":
					doc = result.O
				case "u":
					doc, err = m.getOriginalDoc(result.O2, coll)
					if err != nil { 
						m.pipe.Err <- adaptor.NewError(adaptor.ERROR, m.path, fmt.Sprintf("Mongodb error (%s)", err.Error()), nil)
						continue
					}
				default:
					m.pipe.Err <- adaptor.NewError(adaptor.ERROR, m.path, "Mongodb error (unknown op type)", nil)
					continue
				}
				msg := message.NewMsg(message.OpTypeFromString(result.Op), doc, m.computeNamespace(coll))
				msg.Timestamp = int64(result.Ts) >> 32
				m.oplogTime = result.Ts
				m.pipe.Send(msg)
			}
			result = oplogDoc{}
		}
		if stop := m.pipe.Stopped; stop {
			return
		}
		if iter.Timeout() {
			continue
		}
		if iter.Err() != nil {
			return adaptor.NewError(adaptor.CRITICAL, m.path, fmt.Sprintf("Mongodb error (error reading collection %s)", iter.Err()), nil)
		}
		query = bson.M{
			"ts": bson.M{"$gte": m.oplogTime},
		}
		iter = collection.Find(query).LogReplay().Tail(m.oplogTimeout)
	}
}
func ( m * Mongodb ) tailData ( ) ( err error ) { var ( collection = m . mongoSession . DB ( " " ) . C ( " " ) result oplogDoc query = bson . M { " " : bson . M { " " : m . oplogTime } , } iter = collection . Find ( query ) . LogReplay ( ) . Sort ( " " ) . Tail ( m . oplogTimeout ) ) for { for iter . Next ( & result ) { if stop := m . pipe . Stopped ; stop { return } if result . validOp ( ) { db , coll , _ := m . splitNamespace ( result . Ns ) if db != m . database { continue } if strings . HasPrefix ( coll , " " ) { continue } else if match := m . collectionMatch . MatchString ( coll ) ; ! match { continue } var doc bson . M switch result . Op { case " " : doc = result . O case " " : doc = result . O case " " : doc , err = m . getOriginalDoc ( result . O2 , coll ) if err != nil { m . pipe . Err <- adaptor . NewError ( adaptor . ERROR , m . path , fmt . Sprintf ( " " , err . Error ( ) ) , nil ) continue } default : m . pipe . Err <- adaptor . NewError ( adaptor . ERROR , m . path , " " , nil ) continue } msg := message . NewMsg ( message . OpTypeFromString ( result . Op ) , doc , m . computeNamespace ( coll ) ) msg . Timestamp = int64 ( result . Ts ) >> 32 m . oplogTime = result . Ts m . pipe . Send ( msg ) } result = oplogDoc { } } if stop := m . pipe . Stopped ; stop { return } if iter . Timeout ( ) { continue } if iter . Err ( ) != nil { return adaptor . NewError ( adaptor . CRITICAL , m . path , fmt . Sprintf ( " " , iter . Err ( ) ) , nil ) } query = bson . M { " " : bson . M { " " : m . oplogTime } , } iter = collection . Find ( query ) . LogReplay ( ) . Tail ( m . oplogTimeout ) } }
tail the oplog
tail the oplog
// hold the document // errors aren't fatal here, but we need to send it down the pipe // we've exited the mongo read loop, lets figure out why // check here again if we've been asked to quit // query will change,