oplogc

README

OpLog Consumer

This repository contains a Go library to connect as a consumer to an OpLog server.

Here is an example of a Go consumer using the provided consumer library:

import (
    "fmt"

    "github.com/dailymotion/oplogc"
)

func main() {
    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
        }
    }
}

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.

Licenses

All source code is licensed under the MIT License.

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
		}
	}
}

Index

• Variables
• type Consumer
  • func Subscribe(url string, options Options) *Consumer
  • func (c *Consumer) LastID() string
  • func (c *Consumer) SetLastID(id string)
  • func (c *Consumer) Start() (ops chan Operation, errs chan error, done chan bool)
  • func (c *Consumer) Stop()
• type Filter
• type Operation
  • func (o *Operation) Done()
• type OperationData
• type Options

Examples

• Package

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.

Types

type Consumer

type Consumer struct {
	// contains filtered or unexported fields
}

Consumer holds all the information required to connect to an oplog server

func Subscribe

func Subscribe(url string, options Options) *Consumer

Subscribe creates a Consumer to connect to the given URL.

func (*Consumer) LastID

func (c *Consumer) LastID() string

LastID returns the most advanced acked event id

func (*Consumer) SetLastID

func (c *Consumer) SetLastID(id string)

SetLastID sets the last id to the given value and informs the save go routine

func (*Consumer) Start

func (c *Consumer) Start() (ops chan Operation, errs chan error, done chan bool)

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.

func (*Consumer) Stop

func (c *Consumer) Stop()

Stop instructs the Start() loop to stop

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

func (*Operation) Done

func (o *Operation) Done()

Done must be called once the operation has been processed by the consumer

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