How can i wait for multiple signals on same channel in temporal golang

Community Support
1

I have a slice in my workflow and need to wait until i receive a signal for each element in the slice, on the same signal channel. I tried using the following code but it does not seem to wait on all the messages being received.

	selector := workflow.NewSelector(ctx)
	notificationSignalChan := workflow.GetSignalChannel(ctx, "my-channel")

	for i := 0; i < len(container.Items); i++ {
		if container.Items[i].Status != status.Pending {
			continue
		}

		var expectedNotification events.Notification

		selector.AddReceive(notificationSignalChan, func(c workflow.ReceiveChannel, more bool) {
			// So it has to be explicitly consumed here
			c.Receive(ctx, &expectedNotification)
			idx := slices.IndexFunc(container.Items, func(item *model.Item) bool {
				return item.ID == notification.ItemID
			})
			
			recordedAt := workflow.Now(ctx)
			container.Items[idx].Status = status.Processed
			err = workflow.ExecuteActivity(ctx, activities.OnProcessed, container.Items[idx]).Get(ctx, nil)
			
			if err != nil {
				panic(err)
			}
		})
	}
	
	for i := 0; i < len(container.Items); i++ {
		if container.Items[i].Status != status.Pending {
			continue
		}

		selector.Select(ctx)
	}
2

Why AddReceive for each item in the container if you’re looking up the item based on the signal input? Just have a single signal handler for that single signal. Move your AddReceive out of that first loop and delete that first loop.

3

Changed to the following and it runs but is there a better/proper way of keeping it spinning/listening to the channel until all items have received a message? I am just sleeping and then checking the channel again, is this correct?

selector := workflow.NewSelector(ctx)
notificationSignalChan := workflow.GetSignalChannel(ctx, “my-channel”)

	var expectedNotification events.Notification

	selector.AddReceive(notificationSignalChan, func(c workflow.ReceiveChannel, more bool) {
		// So it has to be explicitly consumed here
		c.Receive(ctx, &expectedNotification)
		idx := slices.IndexFunc(container.Items, func(item *model.Item) bool {
			return item.ID == notification.ItemID
		})
		
		recordedAt := workflow.Now(ctx)
		container.Items[idx].Status = status.Processed
		err = workflow.ExecuteActivity(ctx, activities.OnProcessed, container.Items[idx]).Get(ctx, nil)
		
		if err != nil {
			panic(err)
		}
	})
	
	for {
		selector.Select(ctx)
		// check the status of all items and if they are Processed then break
		if allItemsComplete(payout.PayoutInstructions) {
			break
		} else {
			// keep spinning not yet complete
			workflow.Sleep(ctx, time.Second*5)
		}
	}
4

You don’t have to sleep, it will wait on the Select anyways. And since you are doing only one thing in that select, you might as well just receive from the channel directly. This select and channel are modeled of the normal Go concept of a select and channel.