newline battles continue
This commit is contained in:
bel
538
vendor/go.mongodb.org/mongo-driver/x/mongo/driverlegacy/find.go
generated
vendored
Executable file
538
vendor/go.mongodb.org/mongo-driver/x/mongo/driverlegacy/find.go
generated
vendored
Executable file
@@ -0,0 +1,538 @@
|
||||
// Copyright (C) MongoDB, Inc. 2017-present.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
// not use this file except in compliance with the License. You may obtain
|
||||
// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
package driverlegacy
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"time"
|
||||
|
||||
"errors"
|
||||
|
||||
"go.mongodb.org/mongo-driver/bson/bsoncodec"
|
||||
"go.mongodb.org/mongo-driver/mongo/options"
|
||||
"go.mongodb.org/mongo-driver/mongo/readpref"
|
||||
"go.mongodb.org/mongo-driver/x/bsonx"
|
||||
"go.mongodb.org/mongo-driver/x/bsonx/bsoncore"
|
||||
"go.mongodb.org/mongo-driver/x/mongo/driver/description"
|
||||
"go.mongodb.org/mongo-driver/x/mongo/driver/session"
|
||||
"go.mongodb.org/mongo-driver/x/mongo/driver/topology"
|
||||
"go.mongodb.org/mongo-driver/x/mongo/driver/uuid"
|
||||
"go.mongodb.org/mongo-driver/x/network/command"
|
||||
"go.mongodb.org/mongo-driver/x/network/connection"
|
||||
"go.mongodb.org/mongo-driver/x/network/wiremessage"
|
||||
)
|
||||
|
||||
// Find handles the full cycle dispatch and execution of a find command against the provided
|
||||
// topology.
|
||||
func Find(
|
||||
ctx context.Context,
|
||||
cmd command.Find,
|
||||
topo *topology.Topology,
|
||||
selector description.ServerSelector,
|
||||
clientID uuid.UUID,
|
||||
pool *session.Pool,
|
||||
registry *bsoncodec.Registry,
|
||||
opts ...*options.FindOptions,
|
||||
) (*BatchCursor, error) {
|
||||
|
||||
if cmd.Session != nil && cmd.Session.PinnedServer != nil {
|
||||
selector = cmd.Session.PinnedServer
|
||||
}
|
||||
ss, err := topo.SelectServerLegacy(ctx, selector)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
desc := ss.Description()
|
||||
conn, err := ss.ConnectionLegacy(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
if desc.WireVersion.Max < 4 {
|
||||
return legacyFind(ctx, cmd, registry, ss, conn, opts...)
|
||||
}
|
||||
|
||||
rp, err := getReadPrefBasedOnTransaction(cmd.ReadPref, cmd.Session)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cmd.ReadPref = rp
|
||||
|
||||
// If no explicit session and deployment supports sessions, start implicit session.
|
||||
if cmd.Session == nil && topo.SupportsSessions() {
|
||||
cmd.Session, err = session.NewClientSession(pool, clientID, session.Implicit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
fo := options.MergeFindOptions(opts...)
|
||||
if fo.AllowPartialResults != nil {
|
||||
cmd.Opts = append(cmd.Opts, bsonx.Elem{"allowPartialResults", bsonx.Boolean(*fo.AllowPartialResults)})
|
||||
}
|
||||
if fo.BatchSize != nil {
|
||||
elem := bsonx.Elem{"batchSize", bsonx.Int32(*fo.BatchSize)}
|
||||
cmd.Opts = append(cmd.Opts, elem)
|
||||
cmd.CursorOpts = append(cmd.CursorOpts, elem)
|
||||
|
||||
if fo.Limit != nil && *fo.BatchSize != 0 && *fo.Limit <= int64(*fo.BatchSize) {
|
||||
cmd.Opts = append(cmd.Opts, bsonx.Elem{"singleBatch", bsonx.Boolean(true)})
|
||||
}
|
||||
}
|
||||
if fo.Collation != nil {
|
||||
if desc.WireVersion.Max < 5 {
|
||||
return nil, ErrCollation
|
||||
}
|
||||
collDoc, err := bsonx.ReadDoc(fo.Collation.ToDocument())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cmd.Opts = append(cmd.Opts, bsonx.Elem{"collation", bsonx.Document(collDoc)})
|
||||
}
|
||||
if fo.Comment != nil {
|
||||
cmd.Opts = append(cmd.Opts, bsonx.Elem{"comment", bsonx.String(*fo.Comment)})
|
||||
}
|
||||
if fo.CursorType != nil {
|
||||
switch *fo.CursorType {
|
||||
case options.Tailable:
|
||||
cmd.Opts = append(cmd.Opts, bsonx.Elem{"tailable", bsonx.Boolean(true)})
|
||||
case options.TailableAwait:
|
||||
cmd.Opts = append(cmd.Opts, bsonx.Elem{"tailable", bsonx.Boolean(true)}, bsonx.Elem{"awaitData", bsonx.Boolean(true)})
|
||||
}
|
||||
}
|
||||
if fo.Hint != nil {
|
||||
hintElem, err := interfaceToElement("hint", fo.Hint, registry)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
cmd.Opts = append(cmd.Opts, hintElem)
|
||||
}
|
||||
if fo.Limit != nil {
|
||||
limit := *fo.Limit
|
||||
if limit < 0 {
|
||||
cmd.Opts = append(cmd.Opts, bsonx.Elem{"singleBatch", bsonx.Boolean(true)})
|
||||
limit = -1 * limit
|
||||
}
|
||||
|
||||
cmd.Opts = append(cmd.Opts, bsonx.Elem{"limit", bsonx.Int64(limit)})
|
||||
}
|
||||
if fo.Max != nil {
|
||||
maxElem, err := interfaceToElement("max", fo.Max, registry)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
cmd.Opts = append(cmd.Opts, maxElem)
|
||||
}
|
||||
if fo.MaxAwaitTime != nil {
|
||||
// Specified as maxTimeMS on the in the getMore command and not given in initial find command.
|
||||
cmd.CursorOpts = append(cmd.CursorOpts, bsonx.Elem{"maxTimeMS", bsonx.Int64(int64(*fo.MaxAwaitTime / time.Millisecond))})
|
||||
}
|
||||
if fo.MaxTime != nil {
|
||||
cmd.Opts = append(cmd.Opts, bsonx.Elem{"maxTimeMS", bsonx.Int64(int64(*fo.MaxTime / time.Millisecond))})
|
||||
}
|
||||
if fo.Min != nil {
|
||||
minElem, err := interfaceToElement("min", fo.Min, registry)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
cmd.Opts = append(cmd.Opts, minElem)
|
||||
}
|
||||
if fo.NoCursorTimeout != nil {
|
||||
cmd.Opts = append(cmd.Opts, bsonx.Elem{"noCursorTimeout", bsonx.Boolean(*fo.NoCursorTimeout)})
|
||||
}
|
||||
if fo.OplogReplay != nil {
|
||||
cmd.Opts = append(cmd.Opts, bsonx.Elem{"oplogReplay", bsonx.Boolean(*fo.OplogReplay)})
|
||||
}
|
||||
if fo.Projection != nil {
|
||||
projElem, err := interfaceToElement("projection", fo.Projection, registry)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
cmd.Opts = append(cmd.Opts, projElem)
|
||||
}
|
||||
if fo.ReturnKey != nil {
|
||||
cmd.Opts = append(cmd.Opts, bsonx.Elem{"returnKey", bsonx.Boolean(*fo.ReturnKey)})
|
||||
}
|
||||
if fo.ShowRecordID != nil {
|
||||
cmd.Opts = append(cmd.Opts, bsonx.Elem{"showRecordId", bsonx.Boolean(*fo.ShowRecordID)})
|
||||
}
|
||||
if fo.Skip != nil {
|
||||
cmd.Opts = append(cmd.Opts, bsonx.Elem{"skip", bsonx.Int64(*fo.Skip)})
|
||||
}
|
||||
if fo.Snapshot != nil {
|
||||
cmd.Opts = append(cmd.Opts, bsonx.Elem{"snapshot", bsonx.Boolean(*fo.Snapshot)})
|
||||
}
|
||||
if fo.Sort != nil {
|
||||
sortElem, err := interfaceToElement("sort", fo.Sort, registry)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
cmd.Opts = append(cmd.Opts, sortElem)
|
||||
}
|
||||
|
||||
res, err := cmd.RoundTrip(ctx, desc, conn)
|
||||
if err != nil {
|
||||
closeImplicitSession(cmd.Session)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return NewBatchCursor(bsoncore.Document(res), cmd.Session, cmd.Clock, ss.Server, cmd.CursorOpts...)
|
||||
}
|
||||
|
||||
// legacyFind handles the dispatch and execution of a find operation against a pre-3.2 server.
|
||||
func legacyFind(
|
||||
ctx context.Context,
|
||||
cmd command.Find,
|
||||
registry *bsoncodec.Registry,
|
||||
ss *topology.SelectedServer,
|
||||
conn connection.Connection,
|
||||
opts ...*options.FindOptions,
|
||||
) (*BatchCursor, error) {
|
||||
query := wiremessage.Query{
|
||||
FullCollectionName: cmd.NS.DB + "." + cmd.NS.Collection,
|
||||
}
|
||||
|
||||
fo := options.MergeFindOptions(opts...)
|
||||
optsDoc, err := createLegacyOptionsDoc(fo, registry)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if fo.Projection != nil {
|
||||
projDoc, err := interfaceToDocument(fo.Projection, registry)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
projRaw, err := projDoc.MarshalBSON()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
query.ReturnFieldsSelector = projRaw
|
||||
}
|
||||
if fo.Skip != nil {
|
||||
query.NumberToSkip = int32(*fo.Skip)
|
||||
query.SkipSet = true
|
||||
}
|
||||
// batch size of 1 not possible with OP_QUERY because the cursor will be closed immediately
|
||||
if fo.BatchSize != nil && *fo.BatchSize == 1 {
|
||||
query.NumberToReturn = 2
|
||||
} else {
|
||||
query.NumberToReturn = calculateNumberToReturn(fo)
|
||||
}
|
||||
query.Flags = calculateLegacyFlags(fo)
|
||||
|
||||
query.BatchSize = fo.BatchSize
|
||||
if fo.Limit != nil {
|
||||
i := int32(*fo.Limit)
|
||||
query.Limit = &i
|
||||
}
|
||||
|
||||
// set read preference and/or slaveOK flag
|
||||
desc := ss.SelectedDescription()
|
||||
if slaveOkNeeded(cmd.ReadPref, desc) {
|
||||
query.Flags |= wiremessage.SlaveOK
|
||||
}
|
||||
optsDoc = addReadPref(cmd.ReadPref, desc.Server.Kind, optsDoc)
|
||||
|
||||
if cmd.Filter == nil {
|
||||
cmd.Filter = bsonx.Doc{}
|
||||
}
|
||||
|
||||
// filter must be wrapped in $query if other $modifiers are used
|
||||
var queryDoc bsonx.Doc
|
||||
if len(optsDoc) == 0 {
|
||||
queryDoc = cmd.Filter
|
||||
} else {
|
||||
filterDoc := bsonx.Doc{
|
||||
{"$query", bsonx.Document(cmd.Filter)},
|
||||
}
|
||||
// $query should go first
|
||||
queryDoc = append(filterDoc, optsDoc...)
|
||||
}
|
||||
|
||||
queryRaw, err := queryDoc.MarshalBSON()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
query.Query = queryRaw
|
||||
|
||||
reply, err := roundTripQuery(ctx, query, conn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var cursorLimit int32
|
||||
var cursorBatchSize int32
|
||||
if query.Limit != nil {
|
||||
cursorLimit = int32(*query.Limit)
|
||||
if cursorLimit < 0 {
|
||||
cursorLimit *= -1
|
||||
}
|
||||
}
|
||||
if query.BatchSize != nil {
|
||||
cursorBatchSize = int32(*query.BatchSize)
|
||||
}
|
||||
|
||||
// TODO(GODRIVER-617): When the wiremessage package is updated, we should ensure we can get all
|
||||
// of the documents as a single slice instead of having to reslice.
|
||||
ds := new(bsoncore.DocumentSequence)
|
||||
ds.Style = bsoncore.SequenceStyle
|
||||
for _, doc := range reply.Documents {
|
||||
ds.Data = append(ds.Data, doc...)
|
||||
}
|
||||
|
||||
return NewLegacyBatchCursor(cmd.NS, reply.CursorID, ds, cursorLimit, cursorBatchSize, ss.Server)
|
||||
}
|
||||
|
||||
func createLegacyOptionsDoc(fo *options.FindOptions, registry *bsoncodec.Registry) (bsonx.Doc, error) {
|
||||
var optsDoc bsonx.Doc
|
||||
|
||||
if fo.Collation != nil {
|
||||
return nil, ErrCollation
|
||||
}
|
||||
if fo.Comment != nil {
|
||||
optsDoc = append(optsDoc, bsonx.Elem{"$comment", bsonx.String(*fo.Comment)})
|
||||
}
|
||||
if fo.Hint != nil {
|
||||
hintElem, err := interfaceToElement("$hint", fo.Hint, registry)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
optsDoc = append(optsDoc, hintElem)
|
||||
}
|
||||
if fo.Max != nil {
|
||||
maxElem, err := interfaceToElement("$max", fo.Max, registry)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
optsDoc = append(optsDoc, maxElem)
|
||||
}
|
||||
if fo.MaxTime != nil {
|
||||
optsDoc = append(optsDoc, bsonx.Elem{"$maxTimeMS", bsonx.Int64(int64(*fo.MaxTime / time.Millisecond))})
|
||||
}
|
||||
if fo.Min != nil {
|
||||
minElem, err := interfaceToElement("$min", fo.Min, registry)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
optsDoc = append(optsDoc, minElem)
|
||||
}
|
||||
if fo.ReturnKey != nil {
|
||||
optsDoc = append(optsDoc, bsonx.Elem{"$returnKey", bsonx.Boolean(*fo.ReturnKey)})
|
||||
}
|
||||
if fo.ShowRecordID != nil {
|
||||
optsDoc = append(optsDoc, bsonx.Elem{"$showDiskLoc", bsonx.Boolean(*fo.ShowRecordID)})
|
||||
}
|
||||
if fo.Snapshot != nil {
|
||||
optsDoc = append(optsDoc, bsonx.Elem{"$snapshot", bsonx.Boolean(*fo.Snapshot)})
|
||||
}
|
||||
if fo.Sort != nil {
|
||||
sortElem, err := interfaceToElement("$orderby", fo.Sort, registry)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
optsDoc = append(optsDoc, sortElem)
|
||||
}
|
||||
|
||||
return optsDoc, nil
|
||||
}
|
||||
|
||||
func calculateLegacyFlags(fo *options.FindOptions) wiremessage.QueryFlag {
|
||||
var flags wiremessage.QueryFlag
|
||||
|
||||
if fo.AllowPartialResults != nil {
|
||||
flags |= wiremessage.Partial
|
||||
}
|
||||
if fo.CursorType != nil {
|
||||
switch *fo.CursorType {
|
||||
case options.Tailable:
|
||||
flags |= wiremessage.TailableCursor
|
||||
case options.TailableAwait:
|
||||
flags |= wiremessage.TailableCursor
|
||||
flags |= wiremessage.AwaitData
|
||||
}
|
||||
}
|
||||
if fo.NoCursorTimeout != nil {
|
||||
flags |= wiremessage.NoCursorTimeout
|
||||
}
|
||||
if fo.OplogReplay != nil {
|
||||
flags |= wiremessage.OplogReplay
|
||||
}
|
||||
|
||||
return flags
|
||||
}
|
||||
|
||||
// calculate the number to return for the first find query
|
||||
func calculateNumberToReturn(opts *options.FindOptions) int32 {
|
||||
var numReturn int32
|
||||
var limit int32
|
||||
var batchSize int32
|
||||
|
||||
if opts.Limit != nil {
|
||||
limit = int32(*opts.Limit)
|
||||
}
|
||||
if opts.BatchSize != nil {
|
||||
batchSize = int32(*opts.BatchSize)
|
||||
}
|
||||
|
||||
if limit < 0 {
|
||||
numReturn = limit
|
||||
} else if limit == 0 {
|
||||
numReturn = batchSize
|
||||
} else if limit < batchSize {
|
||||
numReturn = limit
|
||||
} else {
|
||||
numReturn = batchSize
|
||||
}
|
||||
|
||||
return numReturn
|
||||
}
|
||||
|
||||
func slaveOkNeeded(rp *readpref.ReadPref, desc description.SelectedServer) bool {
|
||||
if desc.Kind == description.Single && desc.Server.Kind != description.Mongos {
|
||||
return true
|
||||
}
|
||||
if rp == nil {
|
||||
// assume primary
|
||||
return false
|
||||
}
|
||||
|
||||
return rp.Mode() != readpref.PrimaryMode
|
||||
}
|
||||
|
||||
func addReadPref(rp *readpref.ReadPref, kind description.ServerKind, query bsonx.Doc) bsonx.Doc {
|
||||
if !readPrefNeeded(rp, kind) {
|
||||
return query
|
||||
}
|
||||
|
||||
doc := createReadPref(rp)
|
||||
if doc == nil {
|
||||
return query
|
||||
}
|
||||
|
||||
return query.Append("$readPreference", bsonx.Document(doc))
|
||||
}
|
||||
|
||||
func readPrefNeeded(rp *readpref.ReadPref, kind description.ServerKind) bool {
|
||||
if kind != description.Mongos || rp == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
// simple Primary or SecondaryPreferred is communicated via slaveOk to Mongos.
|
||||
if rp.Mode() == readpref.PrimaryMode || rp.Mode() == readpref.SecondaryPreferredMode {
|
||||
if _, ok := rp.MaxStaleness(); !ok && len(rp.TagSets()) == 0 {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
func createReadPref(rp *readpref.ReadPref) bsonx.Doc {
|
||||
if rp == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
doc := bsonx.Doc{}
|
||||
|
||||
switch rp.Mode() {
|
||||
case readpref.PrimaryMode:
|
||||
doc = append(doc, bsonx.Elem{"mode", bsonx.String("primary")})
|
||||
case readpref.PrimaryPreferredMode:
|
||||
doc = append(doc, bsonx.Elem{"mode", bsonx.String("primaryPreferred")})
|
||||
case readpref.SecondaryPreferredMode:
|
||||
doc = append(doc, bsonx.Elem{"mode", bsonx.String("secondaryPreferred")})
|
||||
case readpref.SecondaryMode:
|
||||
doc = append(doc, bsonx.Elem{"mode", bsonx.String("secondary")})
|
||||
case readpref.NearestMode:
|
||||
doc = append(doc, bsonx.Elem{"mode", bsonx.String("nearest")})
|
||||
}
|
||||
|
||||
sets := make([]bsonx.Val, 0, len(rp.TagSets()))
|
||||
for _, ts := range rp.TagSets() {
|
||||
if len(ts) == 0 {
|
||||
continue
|
||||
}
|
||||
set := bsonx.Doc{}
|
||||
for _, t := range ts {
|
||||
set = append(set, bsonx.Elem{t.Name, bsonx.String(t.Value)})
|
||||
}
|
||||
sets = append(sets, bsonx.Document(set))
|
||||
}
|
||||
if len(sets) > 0 {
|
||||
doc = append(doc, bsonx.Elem{"tags", bsonx.Array(sets)})
|
||||
}
|
||||
if d, ok := rp.MaxStaleness(); ok {
|
||||
doc = append(doc, bsonx.Elem{"maxStalenessSeconds", bsonx.Int32(int32(d.Seconds()))})
|
||||
}
|
||||
|
||||
return doc
|
||||
}
|
||||
|
||||
func roundTripQuery(ctx context.Context, query wiremessage.Query, conn connection.Connection) (wiremessage.Reply, error) {
|
||||
err := conn.WriteWireMessage(ctx, query)
|
||||
if err != nil {
|
||||
if _, ok := err.(command.Error); ok {
|
||||
return wiremessage.Reply{}, err
|
||||
}
|
||||
return wiremessage.Reply{}, command.Error{
|
||||
Message: err.Error(),
|
||||
Labels: []string{command.NetworkError},
|
||||
}
|
||||
}
|
||||
|
||||
wm, err := conn.ReadWireMessage(ctx)
|
||||
if err != nil {
|
||||
if _, ok := err.(command.Error); ok {
|
||||
return wiremessage.Reply{}, err
|
||||
}
|
||||
// Connection errors are transient
|
||||
return wiremessage.Reply{}, command.Error{
|
||||
Message: err.Error(),
|
||||
Labels: []string{command.NetworkError},
|
||||
}
|
||||
}
|
||||
|
||||
reply, ok := wm.(wiremessage.Reply)
|
||||
if !ok {
|
||||
return wiremessage.Reply{}, errors.New("did not receive OP_REPLY response")
|
||||
}
|
||||
|
||||
err = validateOpReply(reply)
|
||||
if err != nil {
|
||||
return wiremessage.Reply{}, err
|
||||
}
|
||||
|
||||
return reply, nil
|
||||
}
|
||||
|
||||
func validateOpReply(reply wiremessage.Reply) error {
|
||||
if int(reply.NumberReturned) != len(reply.Documents) {
|
||||
return command.NewCommandResponseError(command.ReplyDocumentMismatch, nil)
|
||||
}
|
||||
|
||||
if reply.ResponseFlags&wiremessage.QueryFailure == wiremessage.QueryFailure {
|
||||
return command.QueryFailureError{
|
||||
Message: "query failure",
|
||||
Response: reply.Documents[0],
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user