Add interceptors to replayer options

internal/internal_worker.go

@@ -1063,12 +1063,18 @@ type WorkflowReplayerOptions struct {
 	// Optional custom data converter to provide for replay. If not set, the
 	// default converter is used.
 	DataConverter converter.DataConverter
+
+	// Interceptors to apply to the worker. Earlier interceptors wrap later
+	// interceptors.
+	Interceptors []WorkerInterceptor
 }
 
 // NewWorkflowReplayer creates an instance of the WorkflowReplayer.
 func NewWorkflowReplayer(options WorkflowReplayerOptions) (*WorkflowReplayer, error) {
+	registry := newRegistry()
+	registry.interceptors = options.Interceptors
 	return &WorkflowReplayer{
-		registry:      newRegistry(),
+		registry:      registry,
 		dataConverter: options.DataConverter,
 	}, nil
 }

test/integration_test.go

@@ -237,6 +237,10 @@ func (ts *IntegrationTestSuite) SetupTest() {
 		options.WorkerStopTimeout = 10 * time.Second
 	}
 
+	if strings.Contains(ts.T().Name(), "ReplayerWithInterceptor") {
+		options.Interceptors = append(options.Interceptors, &localActivityInterceptor{})
+	}
+
 	ts.worker = worker.New(ts.client, ts.taskQueueName, options)
 	ts.workerStopped = false
 	ts.registerWorkflowsAndActivities(ts.worker)
@@ -2333,6 +2337,59 @@ func (ts *IntegrationTestSuite) TestMutableSideEffects() {
 		workflow.Execution{ID: run.GetID(), RunID: run.GetRunID()}))
 }
 
+type localActivityInterceptor struct{ interceptor.InterceptorBase }
+
+type localActivityWorkflowInterceptor struct {
+	interceptor.WorkflowInboundInterceptorBase
+}
+
+func (l *localActivityInterceptor) InterceptWorkflow(
+	ctx workflow.Context,
+	next interceptor.WorkflowInboundInterceptor,
+) interceptor.WorkflowInboundInterceptor {
+	var ret localActivityWorkflowInterceptor
+	ret.Next = next
+	return &ret
+}
+
+func (l *localActivityWorkflowInterceptor) ExecuteWorkflow(
+	ctx workflow.Context,
+	in *interceptor.ExecuteWorkflowInput,
+) (interface{}, error) {
+	// Execute local activity before running workflow
+	var res int
+	var a Activities
+	actCtx := workflow.WithLocalActivityOptions(ctx, workflow.LocalActivityOptions{ScheduleToCloseTimeout: time.Second})
+	if err := workflow.ExecuteLocalActivity(actCtx, a.Echo, 0, 123).Get(ctx, &res); err != nil {
+		return nil, err
+	} else if res != 123 {
+		return nil, fmt.Errorf("expected 123, got %v", res)
+	}
+	return l.Next.ExecuteWorkflow(ctx, in)
+}
+
+func (ts *IntegrationTestSuite) TestReplayerWithInterceptor() {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	// Do basic test
+	var expected []string
+	run, err := ts.client.ExecuteWorkflow(ctx, ts.startWorkflowOptions("test-replayer-interceptor-"+uuid.New()),
+		ts.workflows.Basic)
+	ts.NoError(err)
+	ts.NoError(run.Get(ctx, &expected))
+	ts.EqualValues(expected, ts.activities.invoked())
+
+	// Now replay it with the interceptor
+	replayer, err := worker.NewWorkflowReplayerWithOptions(worker.WorkflowReplayerOptions{
+		Interceptors: []interceptor.WorkerInterceptor{&localActivityInterceptor{}},
+	})
+	ts.NoError(err)
+	replayer.RegisterWorkflow(ts.workflows.Basic)
+	ts.NoError(replayer.ReplayWorkflowExecution(ctx, ts.client.WorkflowService(), nil, ts.config.Namespace,
+		workflow.Execution{ID: run.GetID(), RunID: run.GetRunID()}))
+}
+
 func (ts *IntegrationTestSuite) registerNamespace() {
 	client, err := client.NewNamespaceClient(client.Options{
 		HostPort:          ts.config.ServiceAddr,