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);
        });
    }