CBG-4605: dcp mode for caching tool

base/dcp_client.go

@@ -15,6 +15,7 @@ import (
 	"fmt"
 	"math"
 	"sync"
+	"testing"
 	"time"
 
 	"github.com/couchbase/gocbcore/v10"
@@ -82,16 +83,20 @@ type DCPClientOptions struct {
 
 func NewDCPClient(ctx context.Context, ID string, callback sgbucket.FeedEventCallbackFunc, options DCPClientOptions, bucket *GocbV2Bucket) (*DCPClient, error) {
 
-	numWorkers := DefaultNumWorkers
-	if options.NumWorkers > 0 {
-		numWorkers = options.NumWorkers
-	}
-
 	numVbuckets, err := bucket.GetMaxVbno()
 	if err != nil {
 		return nil, fmt.Errorf("Unable to determine maxVbNo when creating DCP client: %w", err)
 	}
 
+	return newDCPClientWithForBuckets(ctx, ID, callback, options, bucket, numVbuckets)
+}
+
+func newDCPClientWithForBuckets(ctx context.Context, ID string, callback sgbucket.FeedEventCallbackFunc, options DCPClientOptions, bucket *GocbV2Bucket, numVbuckets uint16) (*DCPClient, error) {
+
+	numWorkers := DefaultNumWorkers
+	if options.NumWorkers > 0 {
+		numWorkers = options.NumWorkers
+	}
 	if options.AgentPriority == gocbcore.DcpAgentPriorityHigh {
 		return nil, fmt.Errorf("sync gateway should not set high priority for DCP feeds")
 	}
@@ -657,3 +662,13 @@ func getLatestVbUUID(failoverLog []gocbcore.FailoverEntry) (vbUUID gocbcore.VbUU
 func (dc *DCPClient) GetMetadataKeyPrefix() string {
 	return dc.metadata.GetKeyPrefix()
 }
+
+// StartWorkersForTest will iterate through dcp workers to start them, to be used for caching testing purposes only.
+func (dc *DCPClient) StartWorkersForTest(t *testing.T) {
+	dc.startWorkers(dc.ctx)
+}
+
+// NewDCPClientForTest is a test-only function to create a DCP client with a specific number of vbuckets.
+func NewDCPClientForTest(ctx context.Context, t *testing.T, ID string, callback sgbucket.FeedEventCallbackFunc, options DCPClientOptions, bucket *GocbV2Bucket, numVbuckets uint16) (*DCPClient, error) {
+	return newDCPClientWithForBuckets(ctx, ID, callback, options, bucket, numVbuckets)
+}

db/util_testing.go

@@ -104,6 +104,12 @@ func (db *DatabaseContext) CallProcessEntry(t *testing.T, ctx context.Context, l
 	db.changeCache.processEntry(ctx, log)
 }
 
+// GetCachedChanges will grab cached changes form channel cache for caching tool, not to be used outside test code.
+func (db *DatabaseContext) GetCachedChanges(t *testing.T, ctx context.Context, chanID channels.ID) ([]*LogEntry, error) {
+	logs, err := db.changeCache.getChannelCache().GetCachedChanges(ctx, chanID)
+	return logs, err
+}
+
 func (db *DatabaseContext) NewDCPCachingCountWaiter(tb testing.TB) *StatWaiter {
 	return db.NewStatWaiter(db.DbStats.Database().DCPCachingCount, tb)
 }
@@ -769,3 +775,14 @@ func GetIndexPartitionCount(t *testing.T, bucket *base.GocbV2Bucket, dsName sgbu
 	require.Failf(t, "index not found", "index %s not found in %+v", indexName, output)
 	return 0
 }
+
+// GetMutationListener retrieves mutation listener form database context, to be used only for testing purposes.
+func (db *DatabaseContext) GetMutationListener(t *testing.T) changeListener {
+	return db.mutationListener
+}
+
+// InitChannel is a test-only function to initialize a channel in the channel cache.
+func (db *DatabaseContext) InitChannel(ctx context.Context, t *testing.T, chanName string) error {
+	_, err := db.channelCache.getSingleChannelCache(ctx, channels.NewID(chanName, base.DefaultCollectionID))
+	return err
+}