func (oplog *OpLog) Tail(lastID LastID, filter Filter, out chan<- GenericEvent, stop <-chan bool) {
	var lastEv GenericEvent
	if lastID != nil {
		if r, ok := lastID.(*ReplicationLastID); ok && r.int64 == 0 {
			out <- &Event{
				ID:    "1",
				Event: "reset",
			}
		}
	}
	done := false
	mu := &sync.RWMutex{}
	isDone := func() bool {
		mu.RLock()
		defer mu.RUnlock()
		return done
	}
	wg := sync.WaitGroup{}
	wg.Add(1)
	go func() {
		defer wg.Done()
		db := oplog.db()
		defer db.Session.Close()
		var iter *mgo.Iter
		defer func() {
			if iter != nil {
				iter.Close()
			}
		}()
		b := backoff.NewExponentialBackOff()
		b.MaxElapsedTime = 0  
		b.Reset()
		var replicationFallbackID LastID
		for {
			var err error
			if i, ok := lastID.(*OperationLastID); ok {
				log.Debug("OPLOG start live updates")
				query := bson.M{}
				filter.apply(&query)
				if i != nil {
					query["_id"] = bson.M{"$gt": i.ObjectId}
				}
				iter = db.C("oplog_ops").Find(query).Sort("$natural").Tail(5 * time.Second)
				operation := Operation{}
				for {
					for iter.Next(&operation) {
						if isDone() {
							return
						}
						if oplog.ObjectURL != "" {
							operation.Data.genRef(oplog.ObjectURL)
						}
						out <- operation
						lastEv = operation
					}
					if iter.Timeout() {
						continue
					}
					break
				}
				if isDone() {
					return
				}
				if iter.Err() != nil {
					log.Warnf("OPLOG tail failed with error, try to reconnect: %s", iter.Err())
				} else if operation.ID == nil {
					log.Debug("OPLOG ops collection is empty, retrying")
					time.Sleep(b.NextBackOff())
					continue
				} else {
					b.Reset()
				}
			} else if i, ok := lastID.(*ReplicationLastID); ok {
				log.Debug("OPLOG start replication")
				if replicationFallbackID, err = oplog.LastID(); err != nil {
					log.Warnf("OPLOG error retriving replication fallback id: %s", err)
					goto retry
				}
				query := bson.M{}
				filter.apply(&query)
				tsClause := bson.M{}
				query["ts"] = tsClause
				if i.int64 > 0 {
					tsClause["$gte"] = i.Time()
				}
				if replicationFallbackID != nil {
					tsClause["$lte"] = replicationFallbackID.Time()
				}
				if !i.fallbackMode {
					query["event"] = "insert"
				}
				for {
					iter = db.C("oplog_states").Find(query).Sort("ts").Limit(oplog.PageSize).Iter()
					c := 0
					object := objectState{}
					for iter.Next(&object) {
						if isDone() {
							return
						}
						if oplog.ObjectURL != "" {
							object.Data.genRef(oplog.ObjectURL)
						}
						out <- object
						lastEv = object
						c++
					}
					if isDone() {
						return
					}
					if iter.Err() != nil {
						log.Warnf("OPLOG replication failed with error, retrying: %s", iter.Err())
						goto retry
					}
					if lastEv != nil && c == oplog.PageSize {
						tsClause["$gte"] = lastEv.GetEventID().Time()
						continue
					}
					break
				}
				liveID := ""  
				if lastEv != nil {
					liveID = lastEv.GetEventID().String()
				}
				out <- &Event{
					ID:    liveID,
					Event: "live",
				}
				lastID = replicationFallbackID
				replicationFallbackID = nil
				lastEv = nil
				b.Reset()
			} else {
				fmt.Printf("%#v", lastID)
				panic("Invalid last id type")
			}
		retry:
			iter.Close()
			time.Sleep(b.NextBackOff())
			db.Session.Refresh()
			if lastEv != nil {
				lastID = lastEv.GetEventID()
			}
		}
	}()
	select {
	case <-stop:
		mu.Lock()
		done = true
		mu.Unlock()
		wg.Wait()
		log.Info("OPLOG tail closed")
	}
}
Tail tails all the new operations in the oplog and send the operation in
the given channel. If the lastID parameter is given, all operation posted after
this event will be returned.

If the lastID is a ReplicationLastID (unix timestamp in milliseconds), the tailing will
start by replicating all the objects last updated after the timestamp.

Giving a lastID of 0 mean replicating all the stored objects before tailing the live updates.

The filter argument can be used to filter on some type of objects or objects with given parrents.

The create, update, delete events are streamed back to the sender thru the out channel
package oplog
import (
	"fmt"
	"sync"
	"time"
	log "github.com/Sirupsen/logrus"
	"github.com/cenkalti/backoff"
	"gopkg.in/mgo.v2"
	"gopkg.in/mgo.v2/bson"
)
type OpLog struct {
	s     *mgo.Session
	Stats *Stats
	ObjectURL string
	PageSize int
}
func New(mongoURL string, maxBytes int) (*OpLog, error) {
	session, err := mgo.Dial(mongoURL)
	if err != nil {
		return nil, err
	}
	session.SetSyncTimeout(10 * time.Second)
	session.SetSocketTimeout(20 * time.Second)
	session.SetSafe(&mgo.Safe{})
	sts := newStats()
	oplog := &OpLog{
		s:        session,
		Stats:    &sts,
		PageSize: 1000,
	}
	oplog.init(maxBytes)
	session.SetMode(mgo.Monotonic, true)
	return oplog, nil
}
func (oplog *OpLog) db() *mgo.Database {
	return oplog.s.Copy().DB("")
}
func (oplog *OpLog) init(maxBytes int) {
	oplogExists := false
	objectsExists := false
	names, _ := oplog.s.DB("").CollectionNames()
	for _, name := range names {
		switch name {
		case "oplog_ops":
			oplogExists = true
		case "oplog_states":
			objectsExists = true
		}
	}
	if !oplogExists {
		log.Info("OPLOG creating capped collection")
		err := oplog.s.DB("").C("oplog_ops").Create(&mgo.CollectionInfo{
			Capped:   true,
			MaxBytes: maxBytes,
		})
		if err != nil {
			log.Fatal(err)
		}
	}
	if !objectsExists {
		log.Info("OPLOG creating objects index")
		c := oplog.s.DB("").C("oplog_states")
		if err := c.EnsureIndexKey("event", "ts"); err != nil {
			log.Fatal(err)
		}
		if err := c.EnsureIndexKey("event", "data.t", "ts"); err != nil {
			log.Fatal(err)
		}
		if err := c.EnsureIndexKey("ts"); err != nil {
			log.Fatal(err)
		}
		if err := c.EnsureIndexKey("data.t", "ts"); err != nil {
			log.Fatal(err)
		}
	}
}
func (oplog *OpLog) Ingest(ops <-chan *Operation, done <-chan bool) {
	db := oplog.db()
	defer db.Session.Close()
	for {
		select {
		case op := <-ops:
			oplog.Stats.QueueSize.Set(int64(len(ops)))
			oplog.append(op, db)
		case <-done:
			return
		}
	}
}
func (oplog *OpLog) Append(op *Operation) {
	oplog.append(op, nil)
}
func (oplog *OpLog) append(op *Operation, db *mgo.Database) {
	if db == nil {
		db = oplog.db()
		defer db.Session.Close()
	}
	log.Debugf("OPLOG ingest operation: %#v", op.Info())
	b := backoff.NewExponentialBackOff()
	b.MaxElapsedTime = 0  
	b.Reset()
	for {
		if err := db.C("oplog_ops").Insert(op); err != nil {
			log.Warnf("OPLOG can't insert operation, retrying: %s", err)
			time.Sleep(b.NextBackOff())
			db.Session.Refresh()
			continue
		}
		break
	}
	event := op.Event
	if event == "update" {
		event = "insert"
	}
	o := objectState{
		ID:        op.Data.GetID(),
		Event:     event,
		Timestamp: time.Now(),
		Data:      op.Data,
	}
	b.Reset()
	for {
		if _, err := db.C("oplog_states").Upsert(bson.M{"_id": o.ID}, o); err != nil {
			log.Warnf("OPLOG can't upsert object, retrying: %s", err)
			time.Sleep(b.NextBackOff())
			db.Session.Refresh()
			continue
		}
		break
	}
	oplog.Stats.EventsIngested.Add(1)
}
func (oplog *OpLog) Diff(createMap map[string]OperationData, updateMap map[string]OperationData, deleteMap map[string]OperationData) error {
	db := oplog.db()
	defer db.Session.Close()
	dumpTime := time.Unix(0, 0)
	for _, obd := range createMap {
		if obd.Timestamp.After(dumpTime) {
			dumpTime = obd.Timestamp
		}
	}
	obs := objectState{}
	iter := db.C("oplog_states").Find(bson.M{}).Iter()
	for iter.Next(&obs) {
		if obs.Event == "deleted" {
			if obd, ok := createMap[obs.ID]; ok {
				if obd.Timestamp.Before(obs.Data.Timestamp) {
					delete(createMap, obs.ID)
				}
			}
		} else {
			if obd, ok := createMap[obs.ID]; ok {
				delete(createMap, obs.ID)
				if obs.Data.Timestamp.Before(obd.Timestamp) {
					updateMap[obs.ID] = obd
				}
			} else {
				if obs.Data.Timestamp.Before(dumpTime) {
					deleteMap[obs.ID] = *obs.Data
					delete(createMap, obs.ID)
				}
			}
		}
	}
	if iter.Err() != nil {
		return iter.Err()
	}
	return nil
}
func (oplog *OpLog) HasID(id LastID) (bool, error) {
	if olid, ok := id.(*OperationLastID); ok {
		db := oplog.db()
		defer db.Session.Close()
		count, err := db.C("oplog_ops").FindId(olid.ObjectId).Count()
		return count != 0, err
	}
	return true, nil
}
func (oplog *OpLog) LastID() (LastID, error) {
	db := oplog.db()
	defer db.Session.Close()
	operation := &Operation{}
	err := db.C("oplog_ops").Find(nil).Sort("-$natural").One(operation)
	if err == mgo.ErrNotFound {
		return nil, nil
	}
	if operation.ID != nil {
		return &OperationLastID{operation.ID}, nil
	}
	return nil, err
}