mongoutils

Documentation

Overview

Package mongoutils contains utilities for working with MongoDB more effectively.

Index

• Constants
• Variables
• func ChangeStreamBackground(ctx context.Context, cs *mongo.ChangeStream) (<-chan ChangeEventResult, bson.Raw, primitive.Timestamp)
• func DatabaseFromClient(client *mongo.Client, dbName string, opts ...*options.DatabaseOptions) *mongo.Database
• func EnsureIndexes(ctx context.Context, coll *mongo.Collection, indexes ...mongo.IndexModel) error
• func GlobalDatabaseOptions() []*options.DatabaseOptions
• func MustRegisterNamespace(db, coll *string)
• func Namespaces() map[string][]string
• func RandomizeNamespaces() (newNamespaces map[string][]string, restore func())
• func RegisterNamespace(db, coll *string) error
• func RegisterUnmanagedNamespace(db, coll string) error
• func SetGlobalDatabaseOptions(opts ...*options.DatabaseOptions)
• func UnmanagedNamespaces() map[string][]string
• type ChangeEvent
• type ChangeEventNamespace
• type ChangeEventOperationType
• type ChangeEventResult
• type ChangeEventTo
• type ChangeEventUpdateDescription

Constants

const (
	ChangeEventOperationTypeInsert       = ChangeEventOperationType("insert")
	ChangeEventOperationTypeDelete       = ChangeEventOperationType("delete")
	ChangeEventOperationTypeReplace      = ChangeEventOperationType("replace")
	ChangeEventOperationTypeUpdate       = ChangeEventOperationType("update")
	ChangeEventOperationTypeDrop         = ChangeEventOperationType("drop")
	ChangeEventOperationTypeRename       = ChangeEventOperationType("rename")
	ChangeEventOperationTypeDropDatabase = ChangeEventOperationType("dropDatabase")
	ChangeEventOperationTypeInvalidate   = ChangeEventOperationType("invalidate")
)

ChangeEvent operation types.

Variables

var ErrChangeStreamInvalidateEvent = errors.New("change stream invalidated")

ErrChangeStreamInvalidateEvent is returned when a change stream is invalidated. When this happens, a corresponding resume token with an "invalidate" event can be used with the StartAfter option in Watch to restart.

var (
	// SecondaryPreferredDatabaseOption is used to have all database and lower operations use secondary preferred.
	SecondaryPreferredDatabaseOption = options.Database().SetReadPreference(readpref.SecondaryPreferred())
)

Functions

func ChangeStreamBackground

func ChangeStreamBackground(ctx context.Context, cs *mongo.ChangeStream) (<-chan ChangeEventResult, bson.Raw, primitive.Timestamp)

ChangeStreamBackground calls Next in a background goroutine that returns a series of events that can be received after the call is done. It will run until the given context is done. Additionally, on the return of this call, the resume token and/or cluster time of the first getMore is returned. The presence of each has its own significance. When the cluster time is persent, it implies that the returned channel will contain an event that happened at that time. The resume token will also be present and refers to that event. Given how the change stream API works though, the cluster time can be used to restart at the time of that event while the resume token can be used to start after that event. For example, if the change stream were used to find an insertion of a document and find all updates after that insertion, you'd utilize the resume token from the channel. Without doing this you can either a) miss events or b) if no more events ever occurred, you may wait forever. Another example is starting a change stream to watch events for a document found/inserted out-of-band of the change stream. In this case you would use the resume token in the return value of this function. The cluster time can be used if the concurrency of the code is such that what consumes the change stream is concurrent with that which produces the change stream (rpc.mongoDBWebRTCCallQueue is one such case). This is frankly more complicated though. Note: It is encouraged your change stream match on the invalidate event for better error handling.

func DatabaseFromClient

func DatabaseFromClient(client *mongo.Client, dbName string, opts ...*options.DatabaseOptions) *mongo.Database

DatabaseFromClient returns the given database from the client.

func EnsureIndexes

func EnsureIndexes(ctx context.Context, coll *mongo.Collection, indexes ...mongo.IndexModel) error

EnsureIndexes ensures that the given indexes are created on the given collection.

func GlobalDatabaseOptions

func GlobalDatabaseOptions() []*options.DatabaseOptions

GlobalDatabaseOptions gets the options to use on all calls to DatabaseFromClient.

func MustRegisterNamespace

func MustRegisterNamespace(db, coll *string)

MustRegisterNamespace ensures the given database and collection can be registered and panics otherwise.

func Namespaces

func Namespaces() map[string][]string

Namespaces returns a copy of all registered namespaces.

func RandomizeNamespaces

func RandomizeNamespaces() (newNamespaces map[string][]string, restore func())

RandomizeNamespaces is a utility to be used by tests to remap all registered namespaces before tests run in order to isolate where test data is stored. The returned restore function should be called after tests are done in order to restore the namespaces to their former state.

func RegisterNamespace

func RegisterNamespace(db, coll *string) error

RegisterNamespace globally registers the given database and collection as in use with MongoDB. It will error if there's a duplicate registration.

func RegisterUnmanagedNamespace

func RegisterUnmanagedNamespace(db, coll string) error

RegisterUnmanagedNamespace registers a namespace that is known of, but is not directly owned by this program. It will not qualify for randomization.

func SetGlobalDatabaseOptions

func SetGlobalDatabaseOptions(opts ...*options.DatabaseOptions)

SetGlobalDatabaseOptions sets the options to use on all calls to DatabaseFromClient.

func UnmanagedNamespaces

func UnmanagedNamespaces() map[string][]string

UnmanagedNamespaces returns a copy of all unmanaged namespaces.

Types

type ChangeEvent

type ChangeEvent struct {
	ID                       bson.RawValue                `bson:"_id"`
	OperationType            ChangeEventOperationType     `bson:"operationType"`
	FullDocument             bson.RawValue                `bson:"fullDocument"`
	NS                       ChangeEventNamespace         `bson:"ns"`
	To                       ChangeEventTo                `bson:"to"`
	DocumentKey              bson.D                       `bson:"documentKey"`
	UpdateDescription        ChangeEventUpdateDescription `bson:"UpdateDescription"`
	ClusterTime              primitive.Timestamp          `bson:"clusterTime"`
	TransactionNumber        uint64                       `bson:"txnNumber"`
	LogicalSessionIdentifier bson.D                       `bson:"lsid"`
}

A ChangeEvent represents all possible fields that a change stream response document can have.

type ChangeEventNamespace

type ChangeEventNamespace struct {
	Database   string `bson:"db"`
	Collection string `bson:"coll"`
}

ChangeEventNamespace is the namespace (database and or collection) affected by the event.

type ChangeEventOperationType

type ChangeEventOperationType string

ChangeEventOperationType is the type of operation that occurred.

type ChangeEventResult

type ChangeEventResult struct {
	Event       *ChangeEvent
	Error       error
	ResumeToken bson.Raw
}

ChangeEventResult represents either an event happening or an error that happened along the way.

type ChangeEventTo

type ChangeEventTo ChangeEventNamespace

ChangeEventTo is used when operationType is rename; This document displays the new name for the ns collection. This document is omitted for all other values of operationType.

type ChangeEventUpdateDescription

type ChangeEventUpdateDescription struct {
	UpdatedFields bson.D   `bson:"updatedFields"`
	RemovedFields []string `bson:"removedFields"`
}

ChangeEventUpdateDescription is a document describing the fields that were updated or removed by the update operation. This document and its fields only appears if the operationType is update.