Best implementation for workflows listening to infrequent signals

Community Support
, ,
1

Hello, new user of temporal here. I want to have a continually running workflow the listens to an infrequent signal (approx once a day). Is it sufficient to have a forever running for loop that waits for the signal and executes an activity in response to a signal?

var mySignal signal
sigChannel := workflow.GetSignalChannel(ctx, mySignal)

for {
sigChannel.Receive(ctx, &mySignal) 
err := workflow.ExecuteActivity(ctx, u.myActivity, mySignal).Get(ctx, nil)
}
return workflow.ContinueAsNew()
3

I think this looks ok. You probably want to return NewContinueAsNewError after X number of received signals and invoked activities.
You also would need to look at draining the signal channel with continueasnew, see post here for more info.

4

Maybe it’s too complicated for the task but

type SomeSignalYouCanReceiveInThisPhase struct {
	Type    string
	Payload string
}

func SomeWorkflow(ctx workflow.Context) error {
	var (
		aa            *Activities
		stopSelecting bool
	)

	selector := workflow.NewSelector(ctx)
	sigChannel := workflow.GetSignalChannel(ctx, "my_signals_chan")

	selector.AddReceive(sigChannel, func(ch workflow.ReceiveChannel, more bool) {
		var mySignal SomeSignalYouCanReceiveInThisPhase
		ch.Receive(ctx, &mySignal)

		switch mySignal.Type {
		case "restart_as_new":
			// you can restart as new by the signal
		case "some_special_type_of_signal":
			_ = workflow.ExecuteActivity(ctx, aa.MyActivity, mySignal.Payload).Get(ctx, nil)
		default:
			// break on some reason maybe
			stopSelecting = true
		}
	})

	for {
		if workflow.GetInfo(ctx).GetCurrentHistoryLength() > 50_000 {
			stopSelecting = true
		}

		if stopSelecting {
			break
		}

		selector.Select(ctx)
	}

	return workflow.NewContinueAsNewError(ctx, SomeWorkflow /* we have no params here */)
}

type Activities struct{}

func (aa *Activities) MyActivity(ctx context.Context, somePayload string) error { return nil }

Or just that

sigChannel := workflow.GetSignalChannel(ctx, "name_of_signal_chan")

	for {
		var mySignal signal
		sigChannel.Receive(ctx, &mySignal)
		_ = workflow.ExecuteActivity(ctx, u.myActivity, mySignal).Get(ctx, nil)

		if workflow.GetInfo(ctx).GetCurrentHistoryLength() > 50_000 {
			break
		}
	}

	return workflow.NewContinueAsNewError(ctx, WF)