Aggregator: use Fork.set_source() instead of Fork.connect()