MessageGenerator example

private readonly array $handlers,
        array $schedules,
        private readonly ClockInterface $clock = new Clock(),
    ) {
        foreach ($schedules as $schedule) {
            $this->addSchedule($schedule);
        }
    }

    public function addSchedule(Schedule $schedule): void
    {
        $this->addMessageGenerator(new MessageGenerator($schedule, 'schedule_'.$this->index++, $this->clock));
    }

    public function addMessageGenerator(MessageGenerator $generator): void
    {
        $this->generators[] = $generator;
    }

    /** * Schedules messages. * * Valid options are: * * sleep (default: 1000000): Time in microseconds to sleep after no messages are found */

class SchedulerTransportFactoryTest extends TestCase
{
    public function testCreateTransport()
    {
        $trigger = $this->createMock(TriggerInterface::class);
        $serializer = $this->createMock(SerializerInterface::class);
        $clock = $this->createMock(ClockInterface::class);

        $defaultRecurringMessage = RecurringMessage::trigger($trigger, (object) ['id' => 'default']);
        $customRecurringMessage = RecurringMessage::trigger($trigger, (object) ['id' => 'custom']);

        $default = new SchedulerTransport(new MessageGenerator(new SomeScheduleProvider([$defaultRecurringMessage]), 'default', $clock));
        $custom = new SchedulerTransport(new MessageGenerator(new SomeScheduleProvider([$customRecurringMessage]), 'custom', $clock));

        $factory = new SchedulerTransportFactory(
            new Container([
                'default' => fn () => new SomeScheduleProvider([$defaultRecurringMessage]),
                'custom' => fn () => new SomeScheduleProvider([$customRecurringMessage]),
            ]),
            $clock,
        );

        $this->assertEquals($default, $factory->createTransport('schedule://default', [], $serializer));
        

        if (false === $scheduleName = parse_url($dsn, \PHP_URL_HOST)) {
            throw new InvalidArgumentException(sprintf('The given Schedule DSN "%s" is invalid.', $dsn));
        }
        if (!$this->scheduleProviders->has($scheduleName)) {
            throw new InvalidArgumentException(sprintf('The schedule "%s" is not found.', $scheduleName));
        }

        /** @var ScheduleProviderInterface $scheduleProvider */
        $scheduleProvider = $this->scheduleProviders->get($scheduleName);

        return new SchedulerTransport(new MessageGenerator($scheduleProvider, $scheduleName, $this->clock));
    }

    public function supports(string $dsn, array $options): bool
    {
        return str_starts_with($dsn, 'schedule://');
    }
}

$clock = $this->createMock(ClockInterface::class);
        $clock->method('now')->willReturnReference($now);

        foreach ($schedule as $i => $s) {
            if (\is_array($s)) {
                $schedule[$i] = $this->createMessage(...$s);
            }
        }
        $schedule = (new Schedule())->add(...$schedule);
        $schedule->stateful(new ArrayAdapter());

        $scheduler = new MessageGenerator($schedule, 'dummy', $clock);

        // Warmup. The first run always returns nothing.         $this->assertSame([], iterator_to_array($scheduler->getMessages(), false));

        foreach ($runs as $time => $expected) {
            $now = self::makeDateTime($time);
            $this->assertSame($expected, iterator_to_array($scheduler->getMessages(), false));
        }
    }

    /** * @dataProvider messagesProvider */