Documentation ¶
Overview ¶
Package oplogc provides an easy to use client interface for the oplog service.
See https://github.com/dailymotion/oplog for more information on oplog.
In case of a connection failure recovery the ack mechanism allows you to handle operations in parallel without loosing track of which operation has been handled.
See cmd/oplog-tail for another usage example.
Example ¶
package main import ( "log" "github.com/dailymotion/oplogc" ) func main() { myOplogURL := "http://oplog.mydomain.com" c := oplogc.Subscribe(myOplogURL, oplogc.Options{}) ops, errs, done := c.Start() for { select { case op := <-ops: // Got the next operation switch op.Event { case "reset": // reset the data store case "live": // put the service back in production default: // Do something with the operation //url := fmt.Sprintf("http://api.domain.com/%s/%s", op.Data.Type, op.Data.ID) //data := MyAPIGetter(url) //MyDataSyncer(data) } // Ack the fact you handled the operation op.Done() case err := <-errs: switch err { case oplogc.ErrAccessDenied, oplogc.ErrWritingState: c.Stop() log.Fatal(err) case oplogc.ErrResumeFailed: log.Print("Resume failed, forcing full replication") c.SetLastID("0") default: log.Print(err) } case <-done: return } } }
Output:
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
var ErrAccessDenied = errors.New("invalid credentials")
ErrAccessDenied is returned by Subscribe when the oplog requires a password different from the one provided in options.
var ErrConnectionClosed = errors.New("connection closed")
ErrConnectionClosed when the SSE stream has closed unexpectedly
var ErrIncompleteEvent = errors.New("incomplete event")
ErrIncompleteEvent is returned when the decoder only recieved a partial event
var ErrInvalidEvent = errors.New("invalid event")
ErrInvalidEvent is returned when the decoder was not able to unmarshal the event
var ErrResumeFailed = errors.New("resume failed")
ErrResumeFailed is returned when the requested last id was not found by the oplog server. This may happen when the last id is very old or size of the oplog capped collection is too small for the load.
When this error happen, the consumer may choose to either ignore the lost events or force a full replication.
var ErrWritingState = errors.New("writing state file failed")
ErrorWritingState is returned when the last processed id can't be written to the state file.
Functions ¶
This section is empty.
Types ¶
type Consumer ¶
type Consumer struct {
// contains filtered or unexported fields
}
Consumer holds all the information required to connect to an oplog server
func (*Consumer) SetLastID ¶
SetLastID sets the last id to the given value and informs the save go routine
func (*Consumer) Start ¶
Start reads the oplog output and send operations back thru the returned ops channel. The caller must then call the Done() method on operation when it has been handled. Failing to call Done() the operations would prevent any resume in case of connection failure or restart of the process.
Any errors are return on the errs channel. In all cases, the Start() method will try to reconnect and/or ignore the error. It is the callers responsability to stop the process loop by calling the Stop() method.
When the loop has ended, a message is sent thru the done channel.
type Filter ¶
type Filter struct { // A list of types to filter on Types []string // A list of parent type/id to filter on Parents []string }
Filter contains arguments to filter the oplog output
type Operation ¶
type Operation struct { // ID holds the operation id used to resume the streaming in case of connection failure. ID string // Event is the kind of operation. It can be insert, update or delete. Event string // Data holds the operation metadata. Data *OperationData // contains filtered or unexported fields }
Operation represents an OpLog operation
type OperationData ¶
type OperationData struct { // ID is the object id. ID string `json:"id"` // Type is the object type. Type string `json:"type"` // Ref contains the URL to fetch to object refered by the operation. This field may // not be present if the oplog server is not configured to generate this field. Ref string `json:"ref,omitempty"` // Timestamp is the time when the operation happened. Timestamp time.Time `json:"timestamp"` // Parents is a list of strings describing the objects related to the object // refered by the operation. Parents []string `json:"parents"` }
OperationData is the data part of the SSE event for the operation.
type Options ¶
type Options struct { // Path of the state file where to persiste the current oplog position. // If empty string, the state is not stored. StateFile string // AllowReplication activates replication if the state file is not found. // When false, a consumer with no state file will only get future operations. AllowReplication bool // Password to access password protected oplog Password string // Proxy to be used to access oplog Proxy string // Filters to apply on the oplog output Filter Filter }
Options is the subscription options
Directories ¶
Path | Synopsis |
---|---|
cmd
|
|
oplog-tail
The oplog-tail command is a example implementation of the Go oplog consumer library.
|
The oplog-tail command is a example implementation of the Go oplog consumer library. |