Added SignalWorkflowSkippingDecision (#753)

Added support for buffering a signal without delivering it to a workflow.
Needed to test async channel draining before completing a workflow.

Author: mfateev

internal/internal_workflow_testsuite.go

@@ -1804,14 +1804,14 @@ func (env *testWorkflowEnvironmentImpl) cancelWorkflow(callback resultHandler) {
 	}, true)
 }
 
-func (env *testWorkflowEnvironmentImpl) signalWorkflow(name string, input interface{}) {
+func (env *testWorkflowEnvironmentImpl) signalWorkflow(name string, input interface{}, startDecisionTask bool) {
 	data, err := encodeArg(env.GetDataConverter(), input)
 	if err != nil {
 		panic(err)
 	}
 	env.postCallback(func() {
 		env.signalHandler(name, data)
-	}, true)
+	}, startDecisionTask)
 }
 
 func (env *testWorkflowEnvironmentImpl) signalWorkflowByID(workflowID, signalName string, input interface{}) error {
@@ -1888,7 +1888,7 @@ func newTestSessionEnvironment(testWorkflowEnvironment *testWorkflowEnvironmentI
 }
 
 func (t *testSessionEnvironmentImpl) SignalCreationResponse(ctx context.Context, sessionID string) error {
-	t.testWorkflowEnvironment.signalWorkflow(sessionID, t.sessionEnvironmentImpl.getCreationResponse())
+	t.testWorkflowEnvironment.signalWorkflow(sessionID, t.sessionEnvironmentImpl.getCreationResponse(), true)
 	return nil
 }
 

internal/internal_workflow_testsuite_test.go

@@ -1968,6 +1968,38 @@ func (s *WorkflowTestSuiteUnitTest) Test_ContextMisuse() {
 	s.Contains(env.GetWorkflowError().Error(), "block on coroutine which is already blocked")
 }
 
+func (s *WorkflowTestSuiteUnitTest) Test_DrainSignalChannel() {
+	workflowFn := func(ctx Context) (string, error) {
+
+		signalCh := GetSignalChannel(ctx, "test-signal")
+		var s1, s2, s3 string
+		signalCh.Receive(ctx, &s1)
+		if !signalCh.ReceiveAsync(&s2) {
+			return "", errors.New("expected signal")
+		}
+		if signalCh.ReceiveAsync(&s3) {
+			return "", errors.New("unexpected signal")
+		}
+		return s1 + s2, nil
+	}
+
+	RegisterWorkflow(workflowFn)
+	env := s.NewTestWorkflowEnvironment()
+
+	env.RegisterDelayedCallback(func() {
+		env.SignalWorkflowSkippingDecision("test-signal", "s1")
+		env.SignalWorkflow("test-signal", "s2")
+	}, time.Minute)
+
+	env.ExecuteWorkflow(workflowFn)
+
+	s.True(env.IsWorkflowCompleted())
+	s.NoError(env.GetWorkflowError())
+	var result string
+	env.GetWorkflowResult(&result)
+	s.Equal("s1s2", result)
+}
+
 func (s *WorkflowTestSuiteUnitTest) Test_ActivityRetry() {
 	attempt1Count := 0
 	activityFailedFn := func(ctx context.Context) (string, error) {

internal/workflow_testsuite.go

@@ -563,7 +563,14 @@ func (t *TestWorkflowEnvironment) CancelWorkflow() {
 
 // SignalWorkflow sends signal to the currently running test workflow.
 func (t *TestWorkflowEnvironment) SignalWorkflow(name string, input interface{}) {
-	t.impl.signalWorkflow(name, input)
+	t.impl.signalWorkflow(name, input, true)
+}
+
+// SignalWorkflowSkippingDecision sends signal to the currently running test workflow without invoking workflow code.
+// Used to test processing of multiple buffered signals before completing workflow.
+// It must be followed by SignalWorkflow, CancelWorkflow or CompleteActivity to force a decision.
+func (t *TestWorkflowEnvironment) SignalWorkflowSkippingDecision(name string, input interface{}) {
+	t.impl.signalWorkflow(name, input, false)
 }
 
 // SignalWorkflowByID sends signal to the currently running test workflow.