redispub

Documentation

Overview

Package redispub reads messages from an input channel and publishes them to redis. It handles deduplicating messages (across multiple running copies of oplogtoredis), and tracking the timestamp of the last message we successfully publishes (so we can pick up from where we left off if oplogtoredis restarts).

Index

• func LastProcessedTimestamp(redisClient redis.UniversalClient, metadataPrefix string) (bson.MongoTimestamp, time.Time, error)
• func PublishStream(client redis.UniversalClient, in <-chan *Publication, opts *PublishOpts, ...)
• type Publication
• type PublishOpts

Functions

func LastProcessedTimestamp

func LastProcessedTimestamp(redisClient redis.UniversalClient, metadataPrefix string) (bson.MongoTimestamp, time.Time, error)

LastProcessedTimestamp returns the timestamp of the last oplog entry that oplogtoredis processed.

It returns both the bson.MongoTimestamp, as well as the time.Time value that timestamp represents (accurate to within 1 second; mongo timestamps only store second resolution)

If oplogtoredis has not processed any messages, returns redis.Nil as an error.

func PublishStream

func PublishStream(client redis.UniversalClient, in <-chan *Publication, opts *PublishOpts, stop <-chan bool)

PublishStream reads Publications from the given channel and publishes them to Redis.

Types

type Publication

type Publication struct {
	// The two channels to send the message to
	CollectionChannel string
	SpecificChannel   string

	// Msg is the message to send.
	Msg []byte

	// OplogTimestamp is the timestamp of the oplog entry. Note that this serves as *both*
	// a monotonically increasing timestamp *and* a unique identifier --
	// see https://docs.mongodb.com/manual/reference/bson-types/#timestamps
	OplogTimestamp bson.MongoTimestamp

	// TxIdx is the index of the operation within a transaction. Used to supplement OplogTimestamp in a transaction.
	TxIdx uint
}

Publication represents a message to be sent to Redis about an oplog entry.

type PublishOpts

type PublishOpts struct {
	FlushInterval    time.Duration
	DedupeExpiration time.Duration
	MetadataPrefix   string
}

PublishOpts are configuration options you can pass to PublishStream.