import "github.com/256dpi/lungo"
Code:
type post struct {
Title string `bson:"title"`
}
// prepare options
opts := Options{
Store: NewMemoryStore(),
}
// open database
client, engine, err := Open(nil, opts)
if err != nil {
panic(err)
}
// ensure engine is closed
defer engine.Close()
// get db
foo := client.Database("foo")
// get collection
bar := foo.Collection("bar")
// insert post
_, err = bar.InsertOne(nil, &post{
Title: "Hello World!",
})
if err != nil {
panic(err)
}
// query posts
csr, err := bar.Find(nil, bson.M{})
if err != nil {
panic(err)
}
// decode posts
var posts []post
err = csr.All(nil, &posts)
if err != nil {
panic(err)
}
// print documents
fmt.Printf("%+v", posts)
Output:
[{Title:Hello World!}]
bucket.go catalog.go client.go collection.go cursor.go database.go engine.go file.go helpers.go indexes.go lungo.go mongo.go result.go session.go store.go stream.go transaction.go utils.go
const (
BucketMarkerStateUploading = "uploading"
BucketMarkerStateUploaded = "uploaded"
BucketMarkerStateDeleted = "deleted"
)The bucket marker states.
const Local = "local"Local is the local database.
ErrEngineClosed is returned if the engine has been closed.
var ErrFileNotFound = gridfs.ErrFileNotFound
ErrFileNotFound is returned if the specified file was not found in the bucket. The value is the same as gridfs.ErrFileNotFound and can be used interchangeably.
ErrInvalidPosition is returned if the resulting position after a seek operation is invalid.
var ErrNoDocuments = mongo.ErrNoDocuments
ErrNoDocuments is returned by SingleResult if not document has been found. The value is the same as mongo.ErrNoDocuments and can be used interchangeably.
ErrSessionEnded is returned if the session has been ended.
Oplog is the handle for the local oplog namespace.
IsUniquenessError returns true if the provided error is generated due to a document failing a unique index constraint.
Open will open a lungo database using the provided store.
WithSession will yield a session context to the provided callback that uses the specified session.
type Bucket struct {
// contains filtered or unexported fields
}Bucket provides access to a GridFS bucket. The type is generally compatible with gridfs.Bucket from the official driver but allows the passing in of a context on all methods. This ways the bucket theoretically supports multi- document transactions. However, it is not recommended to use transactions for large uploads and instead enable the the tracking mode and claim the uploads to ensure operational safety.
func NewBucket(db IDatabase, opts ...*options.BucketOptions) *Bucket
NewBucket creates a bucket using the provided database and options.
ClaimUpload will claim a tracked upload by creating the file and removing the marker.
Cleanup will remove unfinished uploads older than the specified age and all files marked for deletion.
Delete will remove the specified file from the bucket. If the bucket is tracked, only a marker is inserted that will ensure the file and its chunks are deleted during the next cleanup.
DownloadToStream will download the file with the specified id and write its contents to the provided writer.
func (b *Bucket) DownloadToStreamByName(ctx context.Context, name string, w io.Writer, opts ...*options.NameOptions) (int64, error)
DownloadToStreamByName will download the file with the specified name and write its contents to the provided writer.
Drop will drop the files and chunks collection. If the bucket is tracked, the markers collection is also dropped.
EnableTracking will enable a non-standard mode in which in-progress uploads and deletions are tracked by storing a document in an additional "markers" collection. If enabled, uploads can be suspended and resumed later and must be explicitly claimed. All unclaimed uploads and not fully deleted files can be cleaned up.
EnsureIndexes will check if all required indexes exist and create them when needed. Usually, this is done automatically when uploading the first file using a bucket. However, when transactions are used to upload files, the indexes must be created before the first upload as index creation is prohibited during transactions.
func (b *Bucket) Find(ctx context.Context, filter interface{}, opts ...*options.GridFSFindOptions) (ICursor, error)
Find will perform a query on the underlying files collection.
func (b *Bucket) GetChunksCollection(_ context.Context) ICollection
GetChunksCollection returns the collection used for storing chunks.
func (b *Bucket) GetFilesCollection(_ context.Context) ICollection
GetFilesCollection returns the collection used for storing files.
func (b *Bucket) GetMarkersCollection(_ context.Context) ICollection
GetMarkersCollection returns the collection used for storing markers.
OpenDownloadStream will open a download stream for the file with the specified id.
func (b *Bucket) OpenDownloadStreamByName(ctx context.Context, name string, opts ...*options.NameOptions) (*DownloadStream, error)
OpenDownloadStreamByName will open a download stream for the file with the specified name.
func (b *Bucket) OpenUploadStream(ctx context.Context, name string, opts ...*options.UploadOptions) (*UploadStream, error)
OpenUploadStream will open an upload stream for a new file with the provided name.
func (b *Bucket) OpenUploadStreamWithID(ctx context.Context, id interface{}, name string, opts ...*options.UploadOptions) (*UploadStream, error)
OpenUploadStreamWithID will open an upload stream for a new file with the provided id and name.
Rename will rename the file with the specified id to the provided name.
func (b *Bucket) UploadFromStream(ctx context.Context, name string, r io.Reader, opts ...*options.UploadOptions) (primitive.ObjectID, error)
UploadFromStream will upload a new file using the contents read from the provided reader.
func (b *Bucket) UploadFromStreamWithID(ctx context.Context, id interface{}, name string, r io.Reader, opts ...*options.UploadOptions) error
UploadFromStreamWithID will upload a new file using the contents read from the provided reader.
type BucketChunk struct {
ID primitive.ObjectID `bson:"_id"`
File interface{} `bson:"files_id"`
Num int `bson:"n"`
Data []byte `bson:"data"`
}BucketChunk represents a document stored in the bucket "chunks" collection.
type BucketFile struct {
ID interface{} `bson:"_id"`
Length int `bson:"length"`
ChunkSize int `bson:"chunkSize"`
UploadDate time.Time `bson:"uploadDate"`
Filename string `bson:"filename"`
Metadata interface{} `bson:"metadata,omitempty"`
}BucketFile represents a document stored in the bucket "files" collection.
type BucketMarker struct {
ID primitive.ObjectID `bson:"_id"`
File interface{} `bson:"files_id"`
State string `bson:"state"`
Timestamp time.Time `bson:"timestamp"`
Length int `bson:"length"`
ChunkSize int `bson:"chunkSize"`
Filename string `bson:"filename"`
Metadata interface{} `bson:"metadata,omitempty"`
}BucketMarker represents a document stored in the bucket "markers" collection.
type Catalog struct {
Namespaces map[Handle]*mongokit.Collection
}Catalog is the top level object per database that contains all data.
NewCatalog creates and returns a new catalog.
Clone will clone the catalog. Namespaces need to be cloned separately.
type Client struct {
// contains filtered or unexported fields
}Client wraps an Engine to be mongo compatible.
Connect implements the IClient.Connect method.
Database implements the IClient.Database method.
Disconnect implements the IClient.Disconnect method.
func (c *Client) ListDatabaseNames(ctx context.Context, filter interface{}, opts ...*options.ListDatabasesOptions) ([]string, error)
ListDatabaseNames implements the IClient.ListDatabaseNames method.
func (c *Client) ListDatabases(ctx context.Context, filter interface{}, opts ...*options.ListDatabasesOptions) (mongo.ListDatabasesResult, error)
ListDatabases implements the IClient.ListDatabases method.
NumberSessionsInProgress implements the IClient.NumberSessionsInProgress method.
Ping implements the IClient.Ping method.
StartSession implements the IClient.StartSession method.
UseSession implements the IClient.UseSession method.
func (c *Client) UseSessionWithOptions(ctx context.Context, opt *options.SessionOptions, fn func(ISessionContext) error) error
UseSessionWithOptions implements the IClient.UseSessionWithOptions method.
func (c *Client) Watch(_ context.Context, pipeline interface{}, opts ...*options.ChangeStreamOptions) (IChangeStream, error)
Watch implements the IClient.Watch method.
type Collection struct {
// contains filtered or unexported fields
}Collection wraps an Engine to be mongo compatible.
func (c *Collection) Aggregate(context.Context, interface{}, ...*options.AggregateOptions) (ICursor, error)
Aggregate implements the ICollection.Aggregate method.
func (c *Collection) BulkWrite(ctx context.Context, models []mongo.WriteModel, opts ...*options.BulkWriteOptions) (*mongo.BulkWriteResult, error)
BulkWrite implements the ICollection.BulkWrite method.
func (c *Collection) Clone(opts ...*options.CollectionOptions) (ICollection, error)
Clone implements the ICollection.Clone method.
func (c *Collection) CountDocuments(ctx context.Context, filter interface{}, opts ...*options.CountOptions) (int64, error)
CountDocuments implements the ICollection.CountDocuments method.
func (c *Collection) Database() IDatabase
Database implements the ICollection.Database method.
func (c *Collection) DeleteMany(ctx context.Context, filter interface{}, opts ...*options.DeleteOptions) (*mongo.DeleteResult, error)
DeleteMany implements the ICollection.DeleteMany method.
func (c *Collection) DeleteOne(ctx context.Context, filter interface{}, opts ...*options.DeleteOptions) (*mongo.DeleteResult, error)
DeleteOne implements the ICollection.DeleteOne method.
func (c *Collection) Distinct(ctx context.Context, field string, filter interface{}, opts ...*options.DistinctOptions) ([]interface{}, error)
Distinct implements the ICollection.Distinct method.
func (c *Collection) Drop(ctx context.Context) error
Drop implements the ICollection.Drop method.
func (c *Collection) EstimatedDocumentCount(ctx context.Context, opts ...*options.EstimatedDocumentCountOptions) (int64, error)
EstimatedDocumentCount implements the ICollection.EstimatedDocumentCount method.
func (c *Collection) Find(ctx context.Context, filter interface{}, opts ...*options.FindOptions) (ICursor, error)
Find implements the ICollection.Find method.
func (c *Collection) FindOne(ctx context.Context, filter interface{}, opts ...*options.FindOneOptions) ISingleResult
FindOne implements the ICollection.FindOne method.
func (c *Collection) FindOneAndDelete(ctx context.Context, filter interface{}, opts ...*options.FindOneAndDeleteOptions) ISingleResult
FindOneAndDelete implements the ICollection.FindOneAndDelete method.
func (c *Collection) FindOneAndReplace(ctx context.Context, filter, replacement interface{}, opts ...*options.FindOneAndReplaceOptions) ISingleResult
FindOneAndReplace implements the ICollection.FindOneAndReplace method.
func (c *Collection) FindOneAndUpdate(ctx context.Context, filter, update interface{}, opts ...*options.FindOneAndUpdateOptions) ISingleResult
FindOneAndUpdate implements the ICollection.FindOneAndUpdate method.
func (c *Collection) Indexes() IIndexView
Indexes implements the ICollection.Indexes method.
func (c *Collection) InsertMany(ctx context.Context, documents []interface{}, opts ...*options.InsertManyOptions) (*mongo.InsertManyResult, error)
InsertMany implements the ICollection.InsertMany method.
func (c *Collection) InsertOne(ctx context.Context, document interface{}, opts ...*options.InsertOneOptions) (*mongo.InsertOneResult, error)
InsertOne implements the ICollection.InsertOne method.
func (c *Collection) Name() string
Name implements the ICollection.Name method.
func (c *Collection) ReplaceOne(ctx context.Context, filter, replacement interface{}, opts ...*options.ReplaceOptions) (*mongo.UpdateResult, error)
ReplaceOne implements the ICollection.ReplaceOne method.
func (c *Collection) UpdateMany(ctx context.Context, filter, update interface{}, opts ...*options.UpdateOptions) (*mongo.UpdateResult, error)
UpdateMany implements the ICollection.UpdateMany method.
func (c *Collection) UpdateOne(ctx context.Context, filter, update interface{}, opts ...*options.UpdateOptions) (*mongo.UpdateResult, error)
UpdateOne implements the ICollection.UpdateOne method.
func (c *Collection) Watch(_ context.Context, pipeline interface{}, opts ...*options.ChangeStreamOptions) (IChangeStream, error)
Watch implements the ICollection.Watch method.
type Cursor struct {
// contains filtered or unexported fields
}Cursor wraps a list to be mongo compatible.
All implements the ICursor.All method.
Close implements the ICursor.Close method.
Decode implements the ICursor.Decode method.
Err implements the ICursor.Err method.
ID implements the ICursor.ID method.
Next implements the ICursor.Next method.
RemainingBatchLength implements the ICursor.RemainingBatchLength method.
TryNext implements the ICursor.TryNext method.
type Database struct {
// contains filtered or unexported fields
}Database wraps an Engine to be mongo compatible.
func (d *Database) Aggregate(context.Context, interface{}, ...*options.AggregateOptions) (ICursor, error)
Aggregate implements the IDatabase.Aggregate method.
Client implements the IDatabase.Client method.
func (d *Database) Collection(name string, opts ...*options.CollectionOptions) ICollection
Collection implements the IDatabase.Collection method.
func (d *Database) CreateCollection(ctx context.Context, name string, opts ...*options.CreateCollectionOptions) error
CreateCollection implements the IDatabase.CreateCollection method.
func (d *Database) CreateView(_ context.Context, _, _ string, _ interface{}, _ ...*options.CreateViewOptions) error
CreateView implements the IDatabase.CreateView method.
Drop implements the IDatabase.Drop method.
func (d *Database) ListCollectionNames(ctx context.Context, filter interface{}, opts ...*options.ListCollectionsOptions) ([]string, error)
ListCollectionNames implements the IDatabase.ListCollectionNames method.
func (d *Database) ListCollections(ctx context.Context, filter interface{}, opts ...*options.ListCollectionsOptions) (ICursor, error)
ListCollections implements the IDatabase.ListCollections method.
Name implements the IDatabase.Name method.
func (d *Database) ReadConcern() *readconcern.ReadConcern
ReadConcern implements the IDatabase.ReadConcern method.
ReadPreference implements the IDatabase.ReadPreference method.
func (d *Database) RunCommand(context.Context, interface{}, ...*options.RunCmdOptions) ISingleResult
RunCommand implements the IDatabase.RunCommand method.
func (d *Database) RunCommandCursor(context.Context, interface{}, ...*options.RunCmdOptions) (ICursor, error)
RunCommandCursor implements the IDatabase.RunCommandCursor method.
func (d *Database) Watch(_ context.Context, pipeline interface{}, opts ...*options.ChangeStreamOptions) (IChangeStream, error)
Watch implements the IDatabase.Watch method.
func (d *Database) WriteConcern() *writeconcern.WriteConcern
WriteConcern implements the IDatabase.WriteConcern method.
type DownloadStream struct {
// contains filtered or unexported fields
}DownloadStream is used to download a single file.
func (s *DownloadStream) Close() error
Close will close the download stream.
func (s *DownloadStream) GetFile() *BucketFile
GetFile will return the file that is stream is downloading from.
func (s *DownloadStream) Read(buf []uint8) (int, error)
Read will read bytes into the specified buffer from the current position of the read head.
Seek will reposition the read head using the specified values. A resulting position below zero will yield and error while a position beyond the file length will yield EOF on subsequent reads.
func (s *DownloadStream) Skip(skip int64) (int64, error)
Skip will advance the read head by the specified amount of bytes.
type Engine struct {
// contains filtered or unexported fields
}Engine manages the catalog loaded from a store and provides access to it through transactions. Additionally, it also manages streams that subscribe to changes to the catalog.
CreateEngine will create and return an engine with a loaded catalog from the store.
func (e *Engine) Abort(txn *Transaction)
Abort will abort the specified transaction. To ensure a transaction is always released, Abort should be called after finishing any transaction.
Begin will create a new transaction from the current catalog. A locked transaction must be committed or aborted before another transaction can be started. Unlocked transactions serve as a point in time snapshots and can be just be discarded when not being used further.
Catalog will return the currently used catalog. Any modifications to the returned catalog while using the engine results in undefined behaviour.
Close will close the engine.
func (e *Engine) Commit(txn *Transaction) error
Commit will attempt to store the modified catalog and on success replace the current catalog. If an error is returned the transaction has been aborted and become invalid.
func (e *Engine) Watch(handle Handle, pipeline bsonkit.List, resumeAfter, startAfter bsonkit.Doc, startAt *primitive.Timestamp) (*Stream, error)
Watch will return a stream that is able to consume events from the oplog.
type File struct {
Namespaces map[string]FileNamespace `bson:"namespaces"`
}File is a format for storing catalogs in a single structure.
BuildFile will build a new file from the provided catalog.
BuildCatalog will build a new catalog from the file.
type FileIndex struct {
Key bsonkit.Doc `bson:"key"`
Unique bool `bson:"unique"`
Partial bsonkit.Doc `bson:"partial"`
Expiry time.Duration `bson:"expiry"`
}FileIndex is a single index stored in a file.
type FileNamespace struct {
Documents bsonkit.List `bson:"documents"`
Indexes map[string]FileIndex `bson:"indexes"`
}FileNamespace is a single namespace stored in a file.
type FileStore struct {
// contains filtered or unexported fields
}FileStore writes the catalog to a single file on disk.
NewFileStore creates and returns a new file store.
Load will read the catalog from disk and return it. If no file exists at the specified location an empty catalog is returned.
Store will atomically write the catalog to disk.
Handle is a two component identifier for namespaces where the first part is the database and the second the collection.
String will return the string form of the handle.
Validate will validate the handle.
type IChangeStream interface {
Close(context.Context) error
Decode(interface{}) error
Err() error
ID() int64
Next(context.Context) bool
ResumeToken() bson.Raw
TryNext(context.Context) bool
}IChangeStream defines a generic change stream.
type IClient interface {
Connect(context.Context) error
Database(string, ...*options.DatabaseOptions) IDatabase
Disconnect(context.Context) error
ListDatabaseNames(context.Context, interface{}, ...*options.ListDatabasesOptions) ([]string, error)
ListDatabases(context.Context, interface{}, ...*options.ListDatabasesOptions) (mongo.ListDatabasesResult, error)
NumberSessionsInProgress() int
Ping(context.Context, *readpref.ReadPref) error
StartSession(...*options.SessionOptions) (ISession, error)
UseSession(context.Context, func(ISessionContext) error) error
UseSessionWithOptions(context.Context, *options.SessionOptions, func(ISessionContext) error) error
Watch(context.Context, interface{}, ...*options.ChangeStreamOptions) (IChangeStream, error)
}IClient defines a generic client.
Connect will connect to a MongoDB database and return a lungo compatible client.
NewClient will create and return a new client.
type ICollection interface {
Aggregate(context.Context, interface{}, ...*options.AggregateOptions) (ICursor, error)
BulkWrite(context.Context, []mongo.WriteModel, ...*options.BulkWriteOptions) (*mongo.BulkWriteResult, error)
Clone(...*options.CollectionOptions) (ICollection, error)
CountDocuments(context.Context, interface{}, ...*options.CountOptions) (int64, error)
Database() IDatabase
DeleteMany(context.Context, interface{}, ...*options.DeleteOptions) (*mongo.DeleteResult, error)
DeleteOne(context.Context, interface{}, ...*options.DeleteOptions) (*mongo.DeleteResult, error)
Distinct(context.Context, string, interface{}, ...*options.DistinctOptions) ([]interface{}, error)
Drop(context.Context) error
EstimatedDocumentCount(context.Context, ...*options.EstimatedDocumentCountOptions) (int64, error)
Find(context.Context, interface{}, ...*options.FindOptions) (ICursor, error)
FindOne(context.Context, interface{}, ...*options.FindOneOptions) ISingleResult
FindOneAndDelete(context.Context, interface{}, ...*options.FindOneAndDeleteOptions) ISingleResult
FindOneAndReplace(context.Context, interface{}, interface{}, ...*options.FindOneAndReplaceOptions) ISingleResult
FindOneAndUpdate(context.Context, interface{}, interface{}, ...*options.FindOneAndUpdateOptions) ISingleResult
Indexes() IIndexView
InsertMany(context.Context, []interface{}, ...*options.InsertManyOptions) (*mongo.InsertManyResult, error)
InsertOne(context.Context, interface{}, ...*options.InsertOneOptions) (*mongo.InsertOneResult, error)
Name() string
ReplaceOne(context.Context, interface{}, interface{}, ...*options.ReplaceOptions) (*mongo.UpdateResult, error)
UpdateMany(context.Context, interface{}, interface{}, ...*options.UpdateOptions) (*mongo.UpdateResult, error)
UpdateOne(context.Context, interface{}, interface{}, ...*options.UpdateOptions) (*mongo.UpdateResult, error)
Watch(context.Context, interface{}, ...*options.ChangeStreamOptions) (IChangeStream, error)
}ICollection defines a generic collection.
type ICursor interface {
All(context.Context, interface{}) error
Close(context.Context) error
Decode(interface{}) error
Err() error
ID() int64
Next(context.Context) bool
RemainingBatchLength() int
TryNext(context.Context) bool
}ICursor defines a generic cursor.
type IDatabase interface {
Aggregate(context.Context, interface{}, ...*options.AggregateOptions) (ICursor, error)
Client() IClient
Collection(string, ...*options.CollectionOptions) ICollection
CreateCollection(context.Context, string, ...*options.CreateCollectionOptions) error
CreateView(context.Context, string, string, interface{}, ...*options.CreateViewOptions) error
Drop(context.Context) error
ListCollectionNames(context.Context, interface{}, ...*options.ListCollectionsOptions) ([]string, error)
ListCollections(context.Context, interface{}, ...*options.ListCollectionsOptions) (ICursor, error)
Name() string
ReadConcern() *readconcern.ReadConcern
ReadPreference() *readpref.ReadPref
RunCommand(context.Context, interface{}, ...*options.RunCmdOptions) ISingleResult
RunCommandCursor(context.Context, interface{}, ...*options.RunCmdOptions) (ICursor, error)
Watch(context.Context, interface{}, ...*options.ChangeStreamOptions) (IChangeStream, error)
WriteConcern() *writeconcern.WriteConcern
}IDatabase defines a generic database.
type IIndexView interface {
CreateMany(context.Context, []mongo.IndexModel, ...*options.CreateIndexesOptions) ([]string, error)
CreateOne(context.Context, mongo.IndexModel, ...*options.CreateIndexesOptions) (string, error)
DropAll(context.Context, ...*options.DropIndexesOptions) (bson.Raw, error)
DropOne(context.Context, string, ...*options.DropIndexesOptions) (bson.Raw, error)
List(context.Context, ...*options.ListIndexesOptions) (ICursor, error)
}IIndexView defines a generic index view.
type ISession interface {
ID() bson.Raw
AbortTransaction(context.Context) error
AdvanceClusterTime(bson.Raw) error
AdvanceOperationTime(*primitive.Timestamp) error
Client() IClient
ClusterTime() bson.Raw
CommitTransaction(context.Context) error
EndSession(context.Context)
OperationTime() *primitive.Timestamp
StartTransaction(...*options.TransactionOptions) error
WithTransaction(context.Context, func(ISessionContext) (interface{}, error), ...*options.TransactionOptions) (interface{}, error)
}ISession defines a generic session.
ISessionContext defines a generic session context.
type ISingleResult interface {
Decode(interface{}) error
DecodeBytes() (bson.Raw, error)
Err() error
}ISingleResult defines a generic single result
type IndexView struct {
// contains filtered or unexported fields
}IndexView wraps an Engine to be mongo compatible.
func (v *IndexView) CreateMany(ctx context.Context, indexes []mongo.IndexModel, opts ...*options.CreateIndexesOptions) ([]string, error)
CreateMany implements the IIndexView.CreateMany method.
func (v *IndexView) CreateOne(ctx context.Context, index mongo.IndexModel, opts ...*options.CreateIndexesOptions) (string, error)
CreateOne implements the IIndexView.CreateOne method.
func (v *IndexView) DropAll(ctx context.Context, opts ...*options.DropIndexesOptions) (bson.Raw, error)
DropAll implements the IIndexView.DropAll method.
func (v *IndexView) DropOne(ctx context.Context, name string, opts ...*options.DropIndexesOptions) (bson.Raw, error)
DropOne implements the IIndexView.DropOne method.
func (v *IndexView) List(ctx context.Context, opts ...*options.ListIndexesOptions) (ICursor, error)
List implements the IIndexView.List method.
type MemoryStore struct {
// contains filtered or unexported fields
}MemoryStore holds the catalog in memory.
func NewMemoryStore() *MemoryStore
NewMemoryStore creates and returns a new memory store.
func (m MemoryStore) Load() (*Catalog, error)
Load will return the catalog.
func (m MemoryStore) Store(data *Catalog) error
Store will store the catalog.
MongoClient wraps a mongo.Client to be lungo compatible.
func (c *MongoClient) Database(name string, opts ...*options.DatabaseOptions) IDatabase
Database implements the IClient.Database method.
func (c *MongoClient) StartSession(opts ...*options.SessionOptions) (ISession, error)
StartSession implements the IClient.StartSession method.
func (c *MongoClient) UseSession(ctx context.Context, fn func(ISessionContext) error) error
UseSession implements the IClient.UseSession method.
func (c *MongoClient) UseSessionWithOptions(ctx context.Context, opt *options.SessionOptions, fn func(ISessionContext) error) error
UseSessionWithOptions implements the IClient.UseSessionWithOptions method.
func (c *MongoClient) Watch(ctx context.Context, pipeline interface{}, opts ...*options.ChangeStreamOptions) (IChangeStream, error)
Watch implements the IClient.Watch method.
type MongoCollection struct {
*mongo.Collection
// contains filtered or unexported fields
}MongoCollection wraps a mongo.Collection to be lungo compatible.
func (c *MongoCollection) Aggregate(ctx context.Context, pipeline interface{}, opts ...*options.AggregateOptions) (ICursor, error)
Aggregate implements the ICollection.Aggregate method.
func (c *MongoCollection) Clone(opts ...*options.CollectionOptions) (ICollection, error)
Clone implements the ICollection.Clone method.
func (c *MongoCollection) Database() IDatabase
Database implements the ICollection.Database method.
func (c *MongoCollection) Find(ctx context.Context, filter interface{}, opts ...*options.FindOptions) (ICursor, error)
Find implements the ICollection.Find method.
func (c *MongoCollection) FindOne(ctx context.Context, filter interface{}, opts ...*options.FindOneOptions) ISingleResult
FindOne implements the ICollection.FindOne method.
func (c *MongoCollection) FindOneAndDelete(ctx context.Context, filter interface{}, opts ...*options.FindOneAndDeleteOptions) ISingleResult
FindOneAndDelete implements the ICollection.FindOneAndDelete method.
func (c *MongoCollection) FindOneAndReplace(ctx context.Context, filter, replacement interface{}, opts ...*options.FindOneAndReplaceOptions) ISingleResult
FindOneAndReplace implements the ICollection.FindOneAndReplace method.
func (c *MongoCollection) FindOneAndUpdate(ctx context.Context, filter, update interface{}, opts ...*options.FindOneAndUpdateOptions) ISingleResult
FindOneAndUpdate implements the ICollection.FindOneAndUpdate method.
func (c *MongoCollection) Indexes() IIndexView
Indexes implements the ICollection.Indexes method.
func (c *MongoCollection) Watch(ctx context.Context, pipeline interface{}, opts ...*options.ChangeStreamOptions) (IChangeStream, error)
Watch implements the ICollection.Watch method.
MongoDatabase wraps a mongo.Database to be lungo compatible.
func (d *MongoDatabase) Aggregate(ctx context.Context, pipeline interface{}, opts ...*options.AggregateOptions) (ICursor, error)
Aggregate implements the IDatabase.Aggregate method.
func (d *MongoDatabase) Client() IClient
Client implements the IDatabase.Client method.
func (d *MongoDatabase) Collection(name string, opts ...*options.CollectionOptions) ICollection
Collection implements the IDatabase.Collection method.
func (d *MongoDatabase) CreateCollection(ctx context.Context, name string, opts ...*options.CreateCollectionOptions) error
CreateCollection implements the IDatabase.CreateCollection method.
func (d *MongoDatabase) ListCollections(ctx context.Context, filter interface{}, opts ...*options.ListCollectionsOptions) (ICursor, error)
ListCollections implements the IDatabase.ListCollections method.
func (d *MongoDatabase) RunCommand(ctx context.Context, runCommand interface{}, opts ...*options.RunCmdOptions) ISingleResult
RunCommand implements the IDatabase.RunCommand method.
func (d *MongoDatabase) RunCommandCursor(ctx context.Context, filter interface{}, opts ...*options.RunCmdOptions) (ICursor, error)
RunCommandCursor implements the IDatabase.RunCommandCursor method.
func (d *MongoDatabase) Watch(ctx context.Context, pipeline interface{}, opts ...*options.ChangeStreamOptions) (IChangeStream, error)
Watch implements the IDatabase.Watch method.
MongoIndexView wraps a mongo.IndexView to be lungo compatible.
func (m *MongoIndexView) CreateMany(ctx context.Context, models []mongo.IndexModel, opts ...*options.CreateIndexesOptions) ([]string, error)
CreateMany implements the IIndexView.List method.
func (m *MongoIndexView) CreateOne(ctx context.Context, model mongo.IndexModel, opts ...*options.CreateIndexesOptions) (string, error)
CreateOne implements the IIndexView.List method.
func (m *MongoIndexView) DropAll(ctx context.Context, opts ...*options.DropIndexesOptions) (bson.Raw, error)
DropAll implements the IIndexView.List method.
func (m *MongoIndexView) DropOne(ctx context.Context, name string, opts ...*options.DropIndexesOptions) (bson.Raw, error)
DropOne implements the IIndexView.List method.
func (m *MongoIndexView) List(ctx context.Context, opts ...*options.ListIndexesOptions) (ICursor, error)
List implements the IIndexView.List method.
MongoSession wraps a mongo.Session to be lungo compatible.
func (s *MongoSession) AbortTransaction(ctx context.Context) error
AbortTransaction implements the ISession.Client method.
func (s *MongoSession) Client() IClient
Client implements the ISession.Client method.
func (s *MongoSession) CommitTransaction(ctx context.Context) error
CommitTransaction implements the ISession.Client method.
func (s *MongoSession) EndSession(ctx context.Context)
EndSession implements the ISession.Client method.
func (s *MongoSession) WithTransaction(ctx context.Context, fn func(ISessionContext) (interface{}, error), opts ...*options.TransactionOptions) (interface{}, error)
WithTransaction implements the ISession.WithTransaction method.
type MongoSessionContext struct {
context.Context
*MongoSession
}MongoSessionContext wraps a mongo.SessionContext to be lungo compatible.
Opcode defines the type of an operation.
The available opcodes.
Strings returns the opcode name.
type Operation struct {
// The opcode.
Opcode Opcode
// The filter document (replace, update, delete).
Filter bsonkit.Doc
// The insert, update or replacement document.
Document bsonkit.Doc
// The sorting to apply (replace, update, delete).
Sort bsonkit.Doc
// Whether an upsert should be performed (replace, update).
Upsert bool
// The documents to skip (update, delete).
Skip int
// The limit (update, delete).
Limit int
// The array filter conditions (update).
ArrayFilters bsonkit.List
}Operation defines a single operation.
type Options struct {
// The store used by the engine to load and store the catalog.
Store Store
// The interval at which expired documents are removed.
//
// Default: 60s.
ExpireInterval time.Duration
// The function that is called with errors from the expiry goroutine.
ExpireErrors func(error)
// The maximum size of the oplog.
//
// Default: 1000.
MaxOplogSize int
// The maximum age of any oplog entry.
//
// Default: 1h.
MaxOplogAge time.Duration
}Options is used to configure an engine.
type Result struct {
// The list of matched documents.
Matched bsonkit.List
// The list of inserted, replace or updated documents.
Modified bsonkit.List
// The upserted document.
Upserted bsonkit.Doc
// The error that occurred during the operation.
Error error
}Result describes the outcome of an operation.
type Session struct {
// contains filtered or unexported fields
}Session provides a mongo compatible way to handle transactions.
AbortTransaction implements the ISession.AbortTransaction method.
AdvanceClusterTime implements the ISession.AdvanceClusterTime method.
AdvanceOperationTime implements the ISession.AdvanceOperationTime method.
Client implements the ISession.Client method.
ClusterTime implements the ISession.ClusterTime method.
CommitTransaction implements the ISession.CommitTransaction method.
EndSession implements the ISession.EndSession method.
ID implements the ISession.ID method.
OperationTime implements the ISession.OperationTime method.
func (s *Session) StartTransaction(opts ...*options.TransactionOptions) error
StartTransaction implements the ISession.StartTransaction method.
func (s *Session) Transaction() *Transaction
Transaction will return the active transaction or nil if no transaction has been started.
func (s *Session) WithTransaction(ctx context.Context, fn func(ISessionContext) (interface{}, error), opts ...*options.TransactionOptions) (interface{}, error)
WithTransaction implements the ISession.WithTransaction method.
SessionContext provides a mongo compatible session context.
type SingleResult struct {
// contains filtered or unexported fields
}SingleResult wraps a result to be mongo compatible.
func (r *SingleResult) Decode(out interface{}) error
Decode implements the ISingleResult.Decode method.
func (r *SingleResult) DecodeBytes() (bson.Raw, error)
DecodeBytes implements the ISingleResult.DecodeBytes method.
func (r *SingleResult) Err() error
Err implements the ISingleResult.Err method.
Store is the interface that describes storage adapters.
type Stream struct {
// contains filtered or unexported fields
}Stream provides a mongo compatible way to read oplog events.
Close implements the IChangeStream.Close method.
Decode implements the IChangeStream.Decode method.
Err implements the IChangeStream.Err method.
ID implements the IChangeStream.ID method.
Next implements the IChangeStream.Next method.
ResumeToken implements the IChangeStream.ResumeToken method.
TryNext implements the ICursor.TryNext method.
type Transaction struct {
// contains filtered or unexported fields
}Transaction buffers multiple changes to a catalog.
func NewTransaction(catalog *Catalog) *Transaction
NewTransaction creates and returns a new transaction.
Bulk performs the specified operations in one go. If ordered is true the process is aborted on the first error.
func (t *Transaction) Catalog() *Catalog
Catalog will return the modified catalog by the transaction.
func (t *Transaction) Clean(maxSize int, maxAge time.Duration)
Clean will clean the oplog and only keep up to the specified amount of events and delete events that are older than the specified age.
func (t *Transaction) CountDocuments(handle Handle) (int, error)
CountDocuments will return the number of documents in the specified namespace.
func (t *Transaction) Create(handle Handle) error
Create will ensure that a namespace for the provided handle exists.
func (t *Transaction) CreateIndex(handle Handle, name string, config mongokit.IndexConfig) (string, error)
CreateIndex will create the specified index in the specified namespace. It is a no-op if an index with the same name and configuration already exists.
func (t *Transaction) Delete(handle Handle, query, sort bsonkit.Doc, skip, limit int) (*Result, error)
Delete will remove all matching documents from the namespace. Sort, skip and limit may be supplied to modify the result. The returned result will contain the matched documents.
func (t *Transaction) Dirty() bool
Dirty will return whether the transaction contains changes.
func (t *Transaction) Drop(handle Handle) error
Drop will return the namespace with the specified handle from the catalog. If the second part of the handle is empty, it will drop all namespaces matching the first part.
func (t *Transaction) DropIndex(handle Handle, name string) error
DropIndex will drop the specified index in the specified namespace.
func (t *Transaction) Expire() error
Expire will remove documents that are expired due to a TTL index.
func (t *Transaction) Find(handle Handle, query, sort bsonkit.Doc, skip, limit int) (*Result, error)
Find will query documents from a namespace. Sort, skip and limit may be supplied to modify the result. The returned results will contain the matched list of documents.
Insert will insert the specified documents into the namespace. The engine will automatically generate an object id per document if it is missing. If ordered is enabled the operation is aborted on the first error and the result returned. Otherwise, the engine will try to insert all documents. The returned results will contain the inserted documents and potential errors.
ListCollections will return a list of all collections in the specified db.
ListDatabases will return a list of all databases in the catalog.
ListIndexes will return a list of indexes in the specified namespace.
func (t *Transaction) Replace(handle Handle, query, sort, repl bsonkit.Doc, upsert bool) (*Result, error)
Replace will replace the first matching document with the specified replacement document. If upsert is enabled, it will insert the replacement document if it is missing. The returned result will contain the matched and modified or upserted document.
func (t *Transaction) Update(handle Handle, query, sort, update bsonkit.Doc, skip, limit int, upsert bool, arrayFilters bsonkit.List) (*Result, error)
Update will apply the update to all matching document. Sort, skip and limit may be supplied to modify the result. If upsert is enabled, it will extract constant parts of the query and apply the update and insert the document if it is missing. The returned result will contain the matched and modified or upserted document.
type UploadStream struct {
// contains filtered or unexported fields
}UploadStream is used to upload a single file.
func (s *UploadStream) Abort() error
Abort will abort the upload and remove uploaded chunks. If the bucket is tracked it will also remove the potentially created marker. If the abort fails the upload may get cleaned up.
func (s *UploadStream) Close() error
Close will finish the upload and close the stream. If the bucket is tracked the method will not finalize the upload by creating a file. Instead the user should call ClaimUpload as part of a multi-document transaction to safely claim the upload. Until that happens the upload may be cleaned up.
func (s *UploadStream) Resume() (int64, error)
Resume will try to resume a previous tracked upload that has been suspended. It will return the amount of bytes that have already been written.
func (s *UploadStream) Suspend() (int64, error)
Suspend will upload fully buffered chunks and close the stream. The stream may be reopened and resumed later to finish the upload. Until that happens the upload my be cleaned up.
func (s *UploadStream) Write(data []uint8) (int, error)
Write will write the provided data to chunks in the background. If the bucket is tracked and an upload already exists, it must be resumed before writing more data.
| Path | Synopsis |
|---|---|
| bsonkit | |
| dbkit | |
| mongokit |
Package lungo imports 22 packages (graph) and is imported by 3 packages. Updated 2020-08-05. Refresh now. Tools for package owners.