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();
    }
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();
    }
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 ( ) ; }
Calls the given generator periodically and publishes the return value.

@param int      $interval
@param Message  $message
@param callable $generator

@return ExtendedPromiseInterface
Calls the given generator periodically and publishes the return value .