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
}
HasID checks if an operation id is present in the capped collection.
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
}