Awaiting multiple signals in Godot 4

Signals are one of the go-to tools in Godot, implementing the observer pattern in an easy, engine-first way. They’re great when you need to respond to an event that could come at some point in the future:

func _ready():
    some_object.do_something.connect(_on_something_happened)

func _on_something_happened():
    print('Something happened!')

Or if you want to wait for an event to happen before continuing a bit of code:

func attack():
    play_animation()
    
    await animation_complete
    
    resolve_damage()

But if you need to wait for multiple signals to emit, there’s no first-class way to handle things, a problem I ran into on Unto Deepest Depths when I needed to trigger an exploding enemy, which could then trigger other exploding enemies, sometimes simultaneously, and wait for the chain of events to complete before continuing turn resolution.

Each explosion fires a completion signal, so I just needed a way to wait for all of these signals to fire before continuing code execution. To that end, I came up with a small utility class that allows me to wait for an arbitrary number of signals to emit and wraps it all into a simple interface:

class_name SignalGroup
extends RefCounted

signal _all_complete

var _counter: int = 0

func all(signals: Array) -> void:
	_counter = signals.size()
	
	if _counter == 0:
		return
		
	for sig in signals:
		sig.connect(_on_signal_complete, CONNECT_ONE_SHOT)
	
	await _all_complete
	
func _on_signal_complete() -> void:
	_counter -= 1
	if _counter == 0:
		_all_complete.emit()

To use this class, I instantiate an instance of this class, provide the signals to listen for when calling all, and then await the function itself to complete. In practice, it would look something like this:

# Example set of signals where timers are created and code execution pauses until all time out
var signals = [
    get_tree().create_timer(1.0).timeout,
    get_tree().create_timer(2.0).timeout,
    get_tree().create_timer(3.0).timeout
]

var signal_group = SignalGroup.new()
await signal_group.all(signals)
print('All signals have fired')

The logic behind this class is pretty straight-forward. Each signal is subscribed to, using the CONNECT_ONE_SHOT flag so that it is automatically disconnected after the first time it fires, in case its source triggers it more than once for whatever reason, and an internal counter tracks how many signals still need to be fired. When all signals have fired, the counter will be zero and the internal _all_complete signal will fire. As the all function will not complete until this signal has fired, awaiting the function call itself is all that’s needed wait for all provided signals to fire.

It’s not a perfect implementation, though, as if you call all on the same object again, and before all of the previous signals have completed, the counter will reset to the new number of signals while the old signals will still be able to fire and decrement the counter. For my use case, where these objects are used once and thrown away, that’s not a problem, but if you use this in your own project be sure to instantiate a new object every time you need one, or disconnect any old signals before connecting the new ones. You could probably also implement a factory function of some sort and take usage down to a single line, or create an any function that waits for the first of the provided signals to fire, but those ideas are left as an exercise for the reader.