src/Rxnet/RabbitMq/RabbitMq.php
public function produce($data, $headers = [], $exchange = '', $routingKey)
    {
        return Observable::create(function (ObserverInterface $observer) use ($exchange, $data, $headers, $routingKey) {
            return $this->channel()
                ->flatMap(
                    function (Channel $channel) use ($exchange, $data, $headers, $routingKey) {
                        return $this->exchange($exchange, [], $channel)
                            ->produce($data, $routingKey, $headers)
                            ->doOnNext(function () use ($channel) {
                                $channel->close();
                            });
                    }
                )->subscribe($observer);
        });
    }
public function produce($data, $headers = [], $exchange = '', $routingKey)
    {
        return Observable::create(function (ObserverInterface $observer) use ($exchange, $data, $headers, $routingKey) {
            return $this->channel()
                ->flatMap(
                    function (Channel $channel) use ($exchange, $data, $headers, $routingKey) {
                        return $this->exchange($exchange, [], $channel)
                            ->produce($data, $routingKey, $headers)
                            ->doOnNext(function () use ($channel) {
                                $channel->close();
                            });
                    }
                )->subscribe($observer);
        });
    }
public function produce ( $ data , $ headers = [ ] , $ exchange = '' , $ routingKey ) { return Observable :: create ( function ( ObserverInterface $ observer ) use ( $ exchange , $ data , $ headers , $ routingKey ) { return $ this -> channel ( ) -> flatMap ( function ( Channel $ channel ) use ( $ exchange , $ data , $ headers , $ routingKey ) { return $ this -> exchange ( $ exchange , [ ] , $ channel ) -> produce ( $ data , $ routingKey , $ headers ) -> doOnNext ( function ( ) use ( $ channel ) { $ channel -> close ( ) ; } ) ; } ) -> subscribe ( $ observer ) ; } ) ; }
One time produce on dedicated channel and close after
@param $data
@param array $headers
@param string $exchange
@param $routingKey
@return Observable\AnonymousObservable
One time produce on dedicated channel and close after