public function publishPeriodically($interval, Message $message, callable $generator)
    {
        if (!$this->isConnected) {
            return new RejectedPromise(new \LogicException('The client is not connected.'));
        }

        $deferred = new Deferred();

        $this->timer[] = $this->loop->addPeriodicTimer(
            $interval,
            function () use ($message, $generator, $deferred) {
                $this->publish($message->withPayload($generator($message->getTopic())))->then(
                    function ($value) use ($deferred) {
                        $deferred->notify($value);
                    },
                    function (\Exception $e) use ($deferred) {
                        $deferred->reject($e);
                    }
                );
            }
        );

        return $deferred->promise();
    }