Done all

indeedeng · Jul 5, 2023 · 920991e · 920991e
1 parent 55c744e
commit 920991e
Show file tree

Hide file tree

Showing 3 changed files with 52 additions and 13 deletions.
diff --git a/integ/conditional_close_test.go b/integ/conditional_close_test.go
@@ -53,6 +53,11 @@ func TestConditionalForceCompleteOnInternalChannelEmptyWorkflowCadenceContinueAs
 }
 
 func doTestConditionalForceCompleteOnInternalChannelEmptyWorkflow(t *testing.T, backendType service.BackendType, config *iwfidl.WorkflowConfig) {
+	doTestConditionalForceCompleteOnChannelEmptyWorkflow(t, backendType, config, false)
+	doTestConditionalForceCompleteOnChannelEmptyWorkflow(t, backendType, config, true)
+}
+
+func doTestConditionalForceCompleteOnChannelEmptyWorkflow(t *testing.T, backendType service.BackendType, config *iwfidl.WorkflowConfig, useSignalChannel bool) {
 	assertions := assert.New(t)
 	// start test workflow server
 	wfHandler := conditionalClose.NewHandler()
@@ -72,9 +77,13 @@ func doTestConditionalForceCompleteOnInternalChannelEmptyWorkflow(t *testing.T,
 	})
 
 	// start a workflow
-	wfId := conditionalClose.WorkflowType + strconv.Itoa(int(time.Now().UnixNano()))
+	channelType := "_internal_channel_"
+	if useSignalChannel {
+		channelType = "_signal_channel_"
+	}
+	wfId := conditionalClose.WorkflowType + channelType + strconv.Itoa(int(time.Now().UnixNano()))
 	req := apiClient.DefaultApi.ApiV1WorkflowStartPost(context.Background())
-	_, httpResp, err := req.WorkflowStartRequest(iwfidl.WorkflowStartRequest{
+	startReq := iwfidl.WorkflowStartRequest{
 		WorkflowId:             wfId,
 		IwfWorkflowType:        conditionalClose.WorkflowType,
 		WorkflowTimeoutSeconds: 20,
@@ -83,19 +92,33 @@ func doTestConditionalForceCompleteOnInternalChannelEmptyWorkflow(t *testing.T,
 		WorkflowStartOptions: &iwfidl.WorkflowStartOptions{
 			WorkflowConfigOverride: config,
 		},
-	}).Execute()
+	}
+	if useSignalChannel {
+		startReq.StateInput = &iwfidl.EncodedObject{} // this will tell the workflow to use signal
+	}
+
+	_, httpResp, err := req.WorkflowStartRequest(startReq).Execute()
 	panicAtHttpError(err, httpResp)
 
 	// wait for a second so that query handler is ready for executing PRC
 	time.Sleep(time.Second)
 	// invoke RPC to send 1 messages to the internal channel to unblock the waitUntil
 	// then send another two messages
 	reqRpc := apiClient.DefaultApi.ApiV1WorkflowRpcPost(context.Background())
+	reqSignal := apiClient.DefaultApi.ApiV1WorkflowSignalPost(context.Background())
 	for i := 0; i < 3; i++ {
-		_, httpResp, err = reqRpc.WorkflowRpcRequest(iwfidl.WorkflowRpcRequest{
-			WorkflowId: wfId,
-			RpcName:    conditionalClose.RpcPublishInternalChannel,
-		}).Execute()
+		if useSignalChannel {
+			httpResp, err = reqSignal.WorkflowSignalRequest(iwfidl.WorkflowSignalRequest{
+				WorkflowId:        wfId,
+				SignalChannelName: conditionalClose.TestChannelName,
+			}).Execute()
+		} else {
+			_, httpResp, err = reqRpc.WorkflowRpcRequest(iwfidl.WorkflowRpcRequest{
+				WorkflowId: wfId,
+				RpcName:    conditionalClose.RpcPublishInternalChannel,
+			}).Execute()
+		}
+
 		panicAtHttpError(err, httpResp)
 		if i == 0 {
 			// wait for a second so that the workflow is in execute state
@@ -111,11 +134,15 @@ func doTestConditionalForceCompleteOnInternalChannelEmptyWorkflow(t *testing.T,
 	panicAtHttpError(err, httpResp)
 
 	history, _ := wfHandler.GetTestResult()
-	assertions.Equalf(map[string]int64{
+
+	expectMap := map[string]int64{
 		"S1_start":  3,
 		"S1_decide": 3,
-		conditionalClose.RpcPublishInternalChannel: 3,
-	}, history, "rpc test fail, %v", history)
+	}
+	if !useSignalChannel {
+		expectMap[conditionalClose.RpcPublishInternalChannel] = 3
+	}
+	assertions.Equalf(expectMap, history, "rpc test fail, %v", history)
 
 	assertions.Equal(iwfidl.COMPLETED, resp2.GetWorkflowStatus())
 	assertions.Equal(1, len(resp2.GetResults()))

diff --git a/integ/workflow/conditional_close/routers.go b/integ/workflow/conditional_close/routers.go
@@ -132,7 +132,8 @@ func (h *handler) ApiV1WorkflowStateDecide(c *gin.Context) {
 				StateDecision: &iwfidl.StateDecision{
 					NextStates: []iwfidl.StateMovement{
 						{
-							StateId: State1,
+							StateId:    State1,
+							StateInput: req.StateInput,
 						},
 					},
 					ConditionalClose: conditionalClose,

diff --git a/service/interpreter/workflowImpl.go b/service/interpreter/workflowImpl.go
@@ -281,7 +281,8 @@ func checkClosingWorkflow(
 ) (canGoNext, gracefulComplete, forceComplete, forceFail bool, completeOutput *iwfidl.StateCompletionOutput, err error) {
 	if decision.HasConditionalClose() {
 		conditionClose := decision.ConditionalClose
-		if conditionClose.GetConditionalCloseType() == iwfidl.FORCE_COMPLETE_ON_INTERNAL_CHANNEL_EMPTY {
+		if conditionClose.GetConditionalCloseType() == iwfidl.FORCE_COMPLETE_ON_INTERNAL_CHANNEL_EMPTY ||
+			conditionClose.GetConditionalCloseType() == iwfidl.FORCE_COMPLETE_ON_SIGNAL_CHANNEL_EMPTY {
 			// trigger a signal draining so that all the signal/internal channel messages are processed
 			// TODO https://github.com/indeedeng/iwf/issues/289
 			// https://github.com/indeedeng/iwf/issues/290
@@ -290,7 +291,17 @@ func checkClosingWorkflow(
 			// to workaround, user code will have to use persistence locking
 			signalReceiver.DrainAllUnreceivedSignals(ctx)
 
-			if !internalChannel.HasData(conditionClose.GetChannelName()) {
+			conditionMet := false
+			if conditionClose.GetConditionalCloseType() == iwfidl.FORCE_COMPLETE_ON_INTERNAL_CHANNEL_EMPTY &&
+				!internalChannel.HasData(conditionClose.GetChannelName()) {
+				conditionMet = true
+			}
+			if conditionClose.GetConditionalCloseType() == iwfidl.FORCE_COMPLETE_ON_SIGNAL_CHANNEL_EMPTY &&
+				!signalReceiver.HasSignal(conditionClose.GetChannelName()) {
+				conditionMet = true
+			}
+
+			if conditionMet {
 				// condition is met, force complete the workflow
 				forceComplete = true
 				completeOutput = &iwfidl.StateCompletionOutput{