GOCBC-954: Add internal ability to get agent from bucket via collection

Motivation ---------- Transactions operations receive a collection from the user, however due to us building gocbtransactions on top of gocbcore transactions we don't actually route requests through the collection. Instead we need to get the gocbcore agent that the collection is using. Changes ------- Create a BucketInternal type accessible via Bucket. Add an internal IORouter function to BucketInternal. Add a function to get Bucket from Collection. Change-Id: Ife038540921ae8e1039253ef43b9b025c5c398a5 Reviewed-on: http://review.couchbase.org/c/gocb/+/132630 Reviewed-by: Brett Lawson <brett19@gmail.com> Tested-by: Charles Dixon <chvckd@gmail.com>
couchbase · Jul 20, 2020 · 96c3928 · 96c3928
1 parent 683edc2
commit 96c3928
Show file tree

Hide file tree

Showing 19 changed files with 127 additions and 47 deletions.
diff --git a/bucket_internal.go b/bucket_internal.go
@@ -0,0 +1,20 @@
+package gocb
+
+import gocbcore "github.com/couchbase/gocbcore/v9"
+
+// InternalBucket is used for internal functionality.
+// Internal: This should never be used and is not supported.
+type InternalBucket struct {
+	bucket *Bucket
+}
+
+// Internal returns a CollectionInternal.
+// Internal: This should never be used and is not supported.
+func (b *Bucket) Internal() *InternalBucket {
+	return &InternalBucket{bucket: b}
+}
+
+// IORouter returns the collection's internal core router.
+func (ib *InternalBucket) IORouter() (*gocbcore.Agent, error) {
+	return ib.bucket.connectionManager.connection(ib.bucket.Name())
+}
diff --git a/client.go b/client.go
@@ -21,6 +21,7 @@ type connectionManager interface {
 	getHTTPProvider() (httpProvider, error)
 	getDiagnosticsProvider(bucketName string) (diagnosticsProvider, error)
 	getWaitUntilReadyProvider(bucketName string) (waitUntilReadyProvider, error)
+	connection(bucketName string) (*gocbcore.Agent, error)
 	close() error
 }
 
@@ -127,7 +128,7 @@ func (c *stdConnectionMgr) getKvProvider(bucketName string) (kvProvider, error)
 		return nil, errors.New("cluster not yet connected")
 	}
 	agent := c.agentgroup.GetAgent(bucketName)
-	if c.agentgroup == nil {
+	if agent == nil {
 		return nil, errors.New("bucket not yet connected")
 	}
 	return agent, nil
@@ -207,6 +208,18 @@ func (c *stdConnectionMgr) getWaitUntilReadyProvider(bucketName string) (waitUnt
 	return &waitUntilReadyProviderWrapper{provider: agent}, nil
 }
 
+func (c *stdConnectionMgr) connection(bucketName string) (*gocbcore.Agent, error) {
+	if c.agentgroup == nil {
+		return nil, errors.New("cluster not yet connected")
+	}
+
+	agent := c.agentgroup.GetAgent(bucketName)
+	if agent == nil {
+		return nil, errors.New("bucket not yet connected")
+	}
+	return agent, nil
+}
+
 func (c *stdConnectionMgr) close() error {
 	c.lock.Lock()
 	if c.agentgroup == nil {

diff --git a/collection.go b/collection.go
@@ -11,7 +11,7 @@ type kvTimeoutsConfig struct {
 type Collection struct {
 	collectionName string
 	scope          string
-	bucket         string
+	bucket         *Bucket
 
 	timeoutsConfig kvTimeoutsConfig
 
@@ -28,7 +28,7 @@ func newCollection(scope *Scope, collectionName string) *Collection {
 	return &Collection{
 		collectionName: collectionName,
 		scope:          scope.Name(),
-		bucket:         scope.bucketName,
+		bucket:         scope.bucket,
 
 		timeoutsConfig: scope.timeoutsConfig,
 
@@ -52,9 +52,9 @@ func (c *Collection) ScopeName() string {
 	return c.scope
 }
 
-// BucketName returns the name of the bucket to which this collection belongs.
+// Bucket returns the name of the bucket to which this collection belongs.
 // UNCOMMITTED: This API may change in the future.
-func (c *Collection) BucketName() string {
+func (c *Collection) Bucket() *Bucket {
 	return c.bucket
 }
 
@@ -69,3 +69,7 @@ func (c *Collection) startKvOpTrace(operationName string, tracectx requestSpanCo
 		SetTag("couchbase.collection", c.collectionName).
 		SetTag("couchbase.service", "kv")
 }
+
+func (c *Collection) bucketName() string {
+	return c.bucket.Name()
+}
diff --git a/collection_bulk.go b/collection_bulk.go
@@ -221,7 +221,7 @@ func (item *TouchOp) execute(tracectx requestSpanContext, c *Collection, provide
 			if res.MutationToken.VbUUID != 0 {
 				mutTok := &MutationToken{
 					token:      res.MutationToken,
-					bucketName: c.bucket,
+					bucketName: c.bucketName(),
 				}
 				item.Result.mt = mutTok
 			}
@@ -276,7 +276,7 @@ func (item *RemoveOp) execute(tracectx requestSpanContext, c *Collection, provid
 			if res.MutationToken.VbUUID != 0 {
 				mutTok := &MutationToken{
 					token:      res.MutationToken,
-					bucketName: c.bucket,
+					bucketName: c.bucketName(),
 				}
 				item.Result.mt = mutTok
 			}
@@ -345,7 +345,7 @@ func (item *UpsertOp) execute(tracectx requestSpanContext, c *Collection, provid
 			if res.MutationToken.VbUUID != 0 {
 				mutTok := &MutationToken{
 					token:      res.MutationToken,
-					bucketName: c.bucket,
+					bucketName: c.bucketName(),
 				}
 				item.Result.mt = mutTok
 			}
@@ -413,7 +413,7 @@ func (item *InsertOp) execute(tracectx requestSpanContext, c *Collection, provid
 			if res.MutationToken.VbUUID != 0 {
 				mutTok := &MutationToken{
 					token:      res.MutationToken,
-					bucketName: c.bucket,
+					bucketName: c.bucketName(),
 				}
 				item.Result.mt = mutTok
 			}
@@ -483,7 +483,7 @@ func (item *ReplaceOp) execute(tracectx requestSpanContext, c *Collection, provi
 			if res.MutationToken.VbUUID != 0 {
 				mutTok := &MutationToken{
 					token:      res.MutationToken,
-					bucketName: c.bucket,
+					bucketName: c.bucketName(),
 				}
 				item.Result.mt = mutTok
 			}
@@ -538,7 +538,7 @@ func (item *AppendOp) execute(tracectx requestSpanContext, c *Collection, provid
 			if res.MutationToken.VbUUID != 0 {
 				mutTok := &MutationToken{
 					token:      res.MutationToken,
-					bucketName: c.bucket,
+					bucketName: c.bucketName(),
 				}
 				item.Result.mt = mutTok
 			}
@@ -593,7 +593,7 @@ func (item *PrependOp) execute(tracectx requestSpanContext, c *Collection, provi
 			if res.MutationToken.VbUUID != 0 {
 				mutTok := &MutationToken{
 					token:      res.MutationToken,
-					bucketName: c.bucket,
+					bucketName: c.bucketName(),
 				}
 				item.Result.mt = mutTok
 			}
@@ -661,7 +661,7 @@ func (item *IncrementOp) execute(tracectx requestSpanContext, c *Collection, pro
 			if res.MutationToken.VbUUID != 0 {
 				mutTok := &MutationToken{
 					token:      res.MutationToken,
-					bucketName: c.bucket,
+					bucketName: c.bucketName(),
 				}
 				item.Result.mt = mutTok
 			}
@@ -729,7 +729,7 @@ func (item *DecrementOp) execute(tracectx requestSpanContext, c *Collection, pro
 			if res.MutationToken.VbUUID != 0 {
 				mutTok := &MutationToken{
 					token:      res.MutationToken,
-					bucketName: c.bucket,
+					bucketName: c.bucketName(),
 				}
 				item.Result.mt = mutTok
 			}

diff --git a/collection_crud.go b/collection_crud.go
@@ -720,7 +720,7 @@ func (c *Collection) GetAnyReplica(id string, opts *GetAnyReplicaOptions) (docOu
 	if res == nil {
 		return nil, &KeyValueError{
 			InnerError:     ErrDocumentUnretrievable,
-			BucketName:     c.bucket,
+			BucketName:     c.bucketName(),
 			ScopeName:      c.scope,
 			CollectionName: c.collectionName,
 		}

diff --git a/collection_crud_test.go b/collection_crud_test.go
@@ -1431,8 +1431,8 @@ func (suite *IntegrationTestSuite) TestDurabilityGetFromAnyReplica() {
 	testCases := []tCase{
 		{
 			name:   "insertDurabilityMajorityDoc",
-			method: "Insert",
-			args: []interface{}{"insertDurabilityMajorityDoc", doc, &InsertOptions{
+			method: "Upsert",
+			args: []interface{}{"insertDurabilityMajorityDoc", doc, &UpsertOptions{
 				DurabilityLevel: DurabilityLevelMajority,
 			}},
 			expectCas:         true,
@@ -1514,7 +1514,7 @@ func (suite *UnitTestSuite) TestGetErrorCollectionUnknown() {
 		Return(pendingOp, nil)
 
 	col := &Collection{
-		bucket:         "mock",
+		bucket:         &Bucket{bucketName: "mock"},
 		collectionName: "",
 		scope:          "",
 
@@ -1574,7 +1574,7 @@ func (suite *UnitTestSuite) TestGetErrorProperties() {
 		Return(pendingOp, nil)
 
 	col := &Collection{
-		bucket: "mock",
+		bucket: &Bucket{bucketName: "mock"},
 
 		collectionName: "",
 		scope:          "",

diff --git a/collection_test.go b/collection_test.go
@@ -9,7 +9,7 @@ func (suite *UnitTestSuite) TestCollectionName() {
 	s := b.Scope(sName)
 	c := s.Collection(cName)
 
-	suite.Assert().Equal(bName, c.BucketName())
+	suite.Assert().Equal(bName, c.bucketName())
 	suite.Assert().Equal(sName, c.ScopeName())
 	suite.Assert().Equal(cName, c.Name())
 }
@@ -21,7 +21,7 @@ func (suite *UnitTestSuite) TestDefaultScopeCollectionName() {
 	b := suite.bucket(bName, suite.defaultTimeoutConfig(), nil)
 	c := b.Collection(cName)
 
-	suite.Assert().Equal(bName, c.BucketName())
+	suite.Assert().Equal(bName, c.bucketName())
 	suite.Assert().Equal("_default", c.ScopeName())
 	suite.Assert().Equal(cName, c.Name())
 }
@@ -32,7 +32,7 @@ func (suite *UnitTestSuite) TestDefaultScopeDefaultCollectionName() {
 	b := suite.bucket(bName, suite.defaultTimeoutConfig(), nil)
 	c := b.DefaultCollection()
 
-	suite.Assert().Equal(bName, c.BucketName())
+	suite.Assert().Equal(bName, c.bucketName())
 	suite.Assert().Equal("_default", c.ScopeName())
 	suite.Assert().Equal("_default", c.Name())
 }
diff --git a/error_wrapping.go b/error_wrapping.go
@@ -110,7 +110,7 @@ func maybeEnhanceKVErr(err error, bucketName, scopeName, collName, docKey string
 }
 
 func maybeEnhanceCollKVErr(err error, bucket kvProvider, coll *Collection, docKey string) error {
-	return maybeEnhanceKVErr(err, coll.bucket, coll.Name(), coll.ScopeName(), docKey)
+	return maybeEnhanceKVErr(err, coll.bucketName(), coll.Name(), coll.ScopeName(), docKey)
 }
 
 func maybeEnhanceViewError(err error) error {

diff --git a/kvopmanager.go b/kvopmanager.go
@@ -132,7 +132,7 @@ func (m *kvOpManager) ScopeName() string {
 }
 
 func (m *kvOpManager) BucketName() string {
-	return m.parent.bucket
+	return m.parent.bucketName()
 }
 
 func (m *kvOpManager) ValueBytes() []byte {

diff --git a/mock_analyticsProvider_test.go b/mock_analyticsProvider_test.go
diff --git a/mock_connectionManager_test.go b/mock_connectionManager_test.go
diff --git a/mock_diagnosticsProvider_test.go b/mock_diagnosticsProvider_test.go
diff --git a/mock_httpProvider_test.go b/mock_httpProvider_test.go
diff --git a/mock_kvProvider_test.go b/mock_kvProvider_test.go
diff --git a/mock_queryProvider_test.go b/mock_queryProvider_test.go
diff --git a/mock_searchProvider_test.go b/mock_searchProvider_test.go
diff --git a/mock_viewProvider_test.go b/mock_viewProvider_test.go