Create interface and mock for matcher (#6374)

* Create interface and mock for matcher

Author: fimanishi

service/matching/tasklist/interfaces.go

@@ -21,6 +21,7 @@
 // SOFTWARE.
 
 //go:generate mockgen -package $GOPACKAGE -source $GOFILE -destination interfaces_mock.go github.com/uber/cadence/service/matching/tasklist Manager
+//go:generate mockgen -package $GOPACKAGE -source $GOFILE -destination interfaces_mock.go github.com/uber/cadence/service/matching/tasklist TaskMatcher
 package tasklist
 
 import (
@@ -57,4 +58,16 @@ type (
 		GetTaskListKind() types.TaskListKind
 		TaskListID() *Identifier
 	}
+
+	TaskMatcher interface {
+		DisconnectBlockedPollers()
+		Offer(ctx context.Context, task *InternalTask) (bool, error)
+		OfferOrTimeout(ctx context.Context, startT time.Time, task *InternalTask) (bool, error)
+		OfferQuery(ctx context.Context, task *InternalTask) (*types.QueryWorkflowResponse, error)
+		MustOffer(ctx context.Context, task *InternalTask) error
+		Poll(ctx context.Context, isolationGroup string) (*InternalTask, error)
+		PollForQuery(ctx context.Context) (*InternalTask, error)
+		UpdateRatelimit(rps *float64)
+		Rate() float64
+	}
 )

service/matching/tasklist/interfaces_mock.go

@@ -37,10 +37,10 @@ import (
 	"github.com/uber/cadence/service/matching/event"
 )
 
-// TaskMatcher matches a task producer with a task consumer
+// taskMatcherImpl matches a task producer with a task consumer
 // Producers are usually rpc calls from history or taskReader
 // that drains backlog from db. Consumers are the task list pollers
-type TaskMatcher struct {
+type taskMatcherImpl struct {
 	log log.Logger
 	// synchronous task channel to match producer/consumer for any isolation group
 	// tasks having no isolation requirement are added to this channel
@@ -81,7 +81,7 @@ func newTaskMatcher(
 	isolationGroups []string,
 	log log.Logger,
 	tasklist *Identifier,
-	tasklistKind types.TaskListKind) *TaskMatcher {
+	tasklistKind types.TaskListKind) TaskMatcher {
 	dPtr := config.TaskDispatchRPS
 	limiter := quotas.NewRateLimiter(&dPtr, config.TaskDispatchRPSTTL, config.MinTaskThrottlingBurstSize())
 	isolatedTaskC := make(map[string]chan *InternalTask)
@@ -91,7 +91,7 @@ func newTaskMatcher(
 
 	cancelCtx, cancelFunc := context.WithCancel(context.Background())
 
-	return &TaskMatcher{
+	return &taskMatcherImpl{
 		log:           log,
 		limiter:       limiter,
 		scope:         scope,
@@ -108,7 +108,7 @@ func newTaskMatcher(
 }
 
 // DisconnectBlockedPollers gradually disconnects pollers which are blocked on long polling
-func (tm *TaskMatcher) DisconnectBlockedPollers() {
+func (tm *taskMatcherImpl) DisconnectBlockedPollers() {
 	tm.cancelFunc()
 }
 
@@ -141,7 +141,7 @@ func (tm *TaskMatcher) DisconnectBlockedPollers() {
 //   - ratelimit is exceeded (does not apply to query task)
 //   - context deadline is exceeded
 //   - task is matched and consumer returns error in response channel
-func (tm *TaskMatcher) Offer(ctx context.Context, task *InternalTask) (bool, error) {
+func (tm *taskMatcherImpl) Offer(ctx context.Context, task *InternalTask) (bool, error) {
 	startT := time.Now()
 	if !task.IsForwarded() {
 		err := tm.ratelimit(ctx)
@@ -213,15 +213,16 @@ func (tm *TaskMatcher) Offer(ctx context.Context, task *InternalTask) (bool, err
 				task.IsForwarded() { // task came from a child partition
 				// a forwarded backlog task from a child partition, block trying
 				// to match with a poller until ctx timeout
-				return tm.offerOrTimeout(ctx, startT, task)
+				return tm.OfferOrTimeout(ctx, startT, task)
 			}
 		}
 
 		return false, nil
 	}
 }
 
-func (tm *TaskMatcher) offerOrTimeout(ctx context.Context, startT time.Time, task *InternalTask) (bool, error) {
+// OfferOrTimeout offers a task to a poller and blocks until a poller picks up the task or context timeouts
+func (tm *taskMatcherImpl) OfferOrTimeout(ctx context.Context, startT time.Time, task *InternalTask) (bool, error) {
 	select {
 	case tm.getTaskC(task) <- task: // poller picked up the task
 		if task.ResponseC != nil {
@@ -242,7 +243,7 @@ func (tm *TaskMatcher) offerOrTimeout(ctx context.Context, startT time.Time, tas
 // OfferQuery will either match task to local poller or will forward query task.
 // Local match is always attempted before forwarding is attempted. If local match occurs
 // response and error are both nil, if forwarding occurs then response or error is returned.
-func (tm *TaskMatcher) OfferQuery(ctx context.Context, task *InternalTask) (*types.QueryWorkflowResponse, error) {
+func (tm *taskMatcherImpl) OfferQuery(ctx context.Context, task *InternalTask) (*types.QueryWorkflowResponse, error) {
 	select {
 	case tm.queryTaskC <- task:
 		<-task.ResponseC
@@ -278,7 +279,7 @@ func (tm *TaskMatcher) OfferQuery(ctx context.Context, task *InternalTask) (*typ
 
 // MustOffer blocks until a consumer is found to handle this task
 // Returns error only when context is canceled, expired or the ratelimit is set to zero (allow nothing)
-func (tm *TaskMatcher) MustOffer(ctx context.Context, task *InternalTask) error {
+func (tm *taskMatcherImpl) MustOffer(ctx context.Context, task *InternalTask) error {
 	e := event.E{
 		TaskListName: tm.tasklist.GetName(),
 		TaskListType: tm.tasklist.GetType(),
@@ -390,7 +391,7 @@ forLoop:
 // On success, the returned task could be a query task or a regular task
 // Returns ErrNoTasks when context deadline is exceeded
 // Returns ErrMatcherClosed when matching is closed
-func (tm *TaskMatcher) Poll(ctx context.Context, isolationGroup string) (*InternalTask, error) {
+func (tm *taskMatcherImpl) Poll(ctx context.Context, isolationGroup string) (*InternalTask, error) {
 	startT := time.Now()
 	isolatedTaskC, ok := tm.isolatedTaskC[isolationGroup]
 	if !ok && isolationGroup != "" {
@@ -429,7 +430,7 @@ func (tm *TaskMatcher) Poll(ctx context.Context, isolationGroup string) (*Intern
 // PollForQuery blocks until a *query* task is found or context deadline is exceeded
 // Returns ErrNoTasks when context deadline is exceeded
 // Returns ErrMatcherClosed when matching is closed
-func (tm *TaskMatcher) PollForQuery(ctx context.Context) (*InternalTask, error) {
+func (tm *taskMatcherImpl) PollForQuery(ctx context.Context) (*InternalTask, error) {
 	startT := time.Now()
 	// try local match first without blocking until context timeout
 	if task, err := tm.pollNonBlocking(ctx, nil, nil, tm.queryTaskC); err == nil {
@@ -443,7 +444,7 @@ func (tm *TaskMatcher) PollForQuery(ctx context.Context) (*InternalTask, error)
 }
 
 // UpdateRatelimit updates the task dispatch rate
-func (tm *TaskMatcher) UpdateRatelimit(rps *float64) {
+func (tm *taskMatcherImpl) UpdateRatelimit(rps *float64) {
 	if rps == nil {
 		return
 	}
@@ -457,11 +458,11 @@ func (tm *TaskMatcher) UpdateRatelimit(rps *float64) {
 }
 
 // Rate returns the current rate at which tasks are dispatched
-func (tm *TaskMatcher) Rate() float64 {
+func (tm *taskMatcherImpl) Rate() float64 {
 	return float64(tm.limiter.Limit())
 }
 
-func (tm *TaskMatcher) pollOrForward(
+func (tm *taskMatcherImpl) pollOrForward(
 	ctx context.Context,
 	startT time.Time,
 	isolationGroup string,
@@ -546,7 +547,7 @@ func (tm *TaskMatcher) pollOrForward(
 	}
 }
 
-func (tm *TaskMatcher) poll(
+func (tm *taskMatcherImpl) poll(
 	ctx context.Context,
 	startT time.Time,
 	isolatedTaskC <-chan *InternalTask,
@@ -610,7 +611,7 @@ func (tm *TaskMatcher) poll(
 	}
 }
 
-func (tm *TaskMatcher) pollLocalWait(
+func (tm *taskMatcherImpl) pollLocalWait(
 	ctx context.Context,
 	isolatedTaskC <-chan *InternalTask,
 	taskC <-chan *InternalTask,
@@ -670,7 +671,7 @@ func (tm *TaskMatcher) pollLocalWait(
 	}
 }
 
-func (tm *TaskMatcher) pollNonBlocking(
+func (tm *taskMatcherImpl) pollNonBlocking(
 	ctx context.Context,
 	isolatedTaskC <-chan *InternalTask,
 	taskC <-chan *InternalTask,
@@ -736,21 +737,21 @@ func (tm *TaskMatcher) pollNonBlocking(
 	}
 }
 
-func (tm *TaskMatcher) fwdrPollReqTokenC(isolationGroup string) <-chan *ForwarderReqToken {
+func (tm *taskMatcherImpl) fwdrPollReqTokenC(isolationGroup string) <-chan *ForwarderReqToken {
 	if tm.fwdr == nil {
 		return noopForwarderTokenC
 	}
 	return tm.fwdr.PollReqTokenC(isolationGroup)
 }
 
-func (tm *TaskMatcher) fwdrAddReqTokenC() <-chan *ForwarderReqToken {
+func (tm *taskMatcherImpl) fwdrAddReqTokenC() <-chan *ForwarderReqToken {
 	if tm.fwdr == nil {
 		return noopForwarderTokenC
 	}
 	return tm.fwdr.AddReqTokenC()
 }
 
-func (tm *TaskMatcher) ratelimit(ctx context.Context) error {
+func (tm *taskMatcherImpl) ratelimit(ctx context.Context) error {
 	err := tm.limiter.Wait(ctx)
 	if errors.Is(err, clock.ErrCannotWait) {
 		// "err != ctx.Err()" may also be correct, as that would mean "gave up due to context".
@@ -763,11 +764,11 @@ func (tm *TaskMatcher) ratelimit(ctx context.Context) error {
 	return err // nil if success, non-nil if canceled
 }
 
-func (tm *TaskMatcher) isForwardingAllowed() bool {
+func (tm *taskMatcherImpl) isForwardingAllowed() bool {
 	return tm.fwdr != nil
 }
 
-func (tm *TaskMatcher) getTaskC(task *InternalTask) chan<- *InternalTask {
+func (tm *taskMatcherImpl) getTaskC(task *InternalTask) chan<- *InternalTask {
 	taskC := tm.taskC
 	if isolatedTaskC, ok := tm.isolatedTaskC[task.isolationGroup]; ok && task.isolationGroup != "" {
 		taskC = isolatedTaskC