Function DropFlowWorkflow

flow/workflows/drop_flow.go:126–276

(ctx workflow.Context, input *protos.DropFlowInput)

}

func DropFlowWorkflow(ctx workflow.Context, input *protos.DropFlowInput) error {
	activeSignal := model.NoopSignal
	if input.Resync {
		activeSignal = model.ResyncSignal
	} else {
		activeSignal = model.TerminateSignal
	}

	if err := workflow.SetQueryHandler(ctx, shared.CDCFlowStateQuery, func() (cdc_state.CDCFlowWorkflowState, error) {
		state := cdc_state.CDCFlowWorkflowState{DropFlowInput: input}
		if input.Resync {
			state.CurrentFlowStatus = protos.FlowStatus_STATUS_RESYNC
			state.ActiveSignal = activeSignal
		} else {
			state.CurrentFlowStatus = protos.FlowStatus_STATUS_TERMINATING
			state.ActiveSignal = activeSignal
		}
		return state, nil
	}); err != nil {
		return fmt.Errorf("failed to set `%s` query handler: %w", shared.CDCFlowStateQuery, err)
	}

	logger := workflow.GetLogger(ctx)

	finalDrop := false
	if input.Resync {
		flowSignalChan := model.FlowSignal.GetSignalChannel(ctx)
		flowSignalStateChangeChan := model.FlowSignalStateChange.GetSignalChannel(ctx)
		workflow.Go(ctx, func(gCtx workflow.Context) {
			sigSelector := workflow.NewNamedSelector(gCtx, input.FlowJobName+"-drop-signals")
			sigSelector.AddReceive(gCtx.Done(), func(_ workflow.ReceiveChannel, _ bool) {})

			flowSignalChan.AddToSelector(sigSelector, func(val model.CDCFlowSignal, _ bool) {
				prev := activeSignal
				activeSignal = model.FlowSignalHandler(activeSignal, val, logger)
				if prev != model.TerminateSignal && activeSignal == model.TerminateSignal {
					finalDrop = true
				}
			})

			flowSignalStateChangeChan.AddToSelector(sigSelector, func(req *protos.FlowStateChangeRequest, _ bool) {
				if req.RequestedFlowState == protos.FlowStatus_STATUS_TERMINATING {
					activeSignal = model.TerminateSignal
					input.DropFlowStats = req.DropMirrorStats
					input.SkipDestinationDrop = req.SkipDestinationDrop
					finalDrop = true
				}
			})

			for gCtx.Err() == nil {
				sigSelector.Select(gCtx)
			}
		})
	}

	status := protos.FlowStatus_STATUS_TERMINATING
	if input.Resync {
		status = protos.FlowStatus_STATUS_RESYNC