It returns a cached exchange or a new one if none exists.

@param string $exchangeName

@return AMQPExchange
510923
private function produceExchangeInstance($exchangeName)
    {
        if (isset($this->declaredExchangeInstances[$exchangeName])) {
            return $this->declaredExchangeInstances[$exchangeName];
        }

        if (isset($this->usedExchangeInstances[$exchangeName])) {
            return $this->usedExchangeInstances[$exchangeName];
        }

        $this->ensureChannel();
        $exchange = new AMQPExchange($this->channel);
        $exchange->setName($exchangeName);
        $this->usedExchangeInstances[$exchangeName] = $exchange;
        return $exchange;
    }
608207
public function bindExchange($destinationName, $sourceName, $routingKey = '')
    {
        /**
         * Amqp 1.4.0 will throw when trying to bind exchanges without a routing key even if they are of type fanout.
         * Later versions do not, but upgrading from 1.4.0 would result in a duplication of bindings (there would
         * be one with a routing key and one without), so in order to avoid that we use a routing key for all versions.
         */
        if ($routingKey == '') {
            $routingKey = 'making-sure-there-is-a-routing-key';
        }

        $exchange = $this->produceExchangeInstance($destinationName);
        $exchange->bind($sourceName, $routingKey);
    }
955771
public function get(string $queueName): ?\AMQPEnvelope
    {
        if ($this->shouldSetup()) {
            $this->setup();
        }

        try {
            if (false !== $message = $this->queue($queueName)->get()) {
                return $message;
            }
        } catch (\AMQPQueueException $e) {
            if (404 === $e->getCode() && $this->shouldSetup()) {
                // If we get a 404 for the queue, it means we need to setup the exchange & queue.
                $this->setup();

                return $this->get();
            }

            throw $e;
        }

        return null;
    }
935632
public function deleteExchange($name, array $options = [], callable $callback = null)
    {
        $ifUnused = array_key_exists('ifUnused', $options) ? (bool)$options['ifUnused'] : null;
        $noWait = array_key_exists('noWait', $options) ? (bool)$options['noWait'] : null;

        $outputFrame = Exchange\ExchangeDeleteFrame::create($name, $ifUnused, $noWait);
        $outputFrame->frameChannelId = $this->id;
        $this->connection->command($outputFrame);

        if (is_callable($callback)) {
            $this->on(self::EVENT_ON_CHANNEL_DELETE_EXCHANGE_CALLBACK, $callback);
        }
    }
1284304
def exchange(self, pubkey):
        '''
        
        '''
        try:
            return self.priv.exchange(c_ec.ECDH(), pubkey.publ)
        except ValueError as e:
            raise s_exc.BadEccExchange(mesg=str(e))
327541
public function exchangeRefreshToken($refreshToken)
    {
        return $this->client->exchangeRefreshToken(
            $refreshToken,
            $this->clientId,
            $this->clientSecret
        );
    }
725438
public function getExchangeRate(ExchangeRateQuery $exchangeQuery): ExchangeRateContract
    {
        $currencyPair = $exchangeQuery->getCurrencyPair();

        $response = $this->request(
            sprintf(
                self::LATEST_URL,
                strtolower($currencyPair->getBaseCurrency()),
                strtolower($currencyPair->getQuoteCurrency())
            )
        );

        $data = StringUtil::jsonToArray($response);

        if (!$data['success']) {
            $message = !empty($data['error']) ? $data['error'] : 'Unknown error';

            throw new Exception($message);
        }

        $date = (new \DateTime())->setTimestamp($data['timestamp']);

        return $this->createRate($currencyPair, (float) ($data['ticker']['price']), $date);
    }
737918
public function createContext(array $options, MessageInterface $message = null, Exchange $exchange = null, \Exception $exception = null)
    {
        $context = [
            self::CONTEXT_OPTIONS => $options,
            self::CONTEXT_VARS => [],
            self::CONTEXT_RESULTS => [],
        ];

        if ($exchange) {
            $context[self::CONTEXT_EXCHANGE] = $exchange;
            if (!$message) {
                $message = $exchange->getIn();
            }
        }

        if ($message) {
            $context[self::CONTEXT_MSG] = $message;
            $context[self::CONTEXT_HEADERS] = $message->getHeaders();
            $context[self::CONTEXT_BODY] = $message->getBody();
        }

        if ($exception) {
            $context[self::CONTEXT_EXCEPTION] = $exception;
        }

        return $context;
    }
994730
protected Request getRequest(HttpServerExchange exchange) {
        final String authenticity = Optional.ofNullable(this.attachment.getRequestParameter()
                .get(Default.AUTHENTICITY.toString()))
                .orElse(this.attachment.getForm().get(Default.AUTHENTICITY.toString()));
        
        return new Request(exchange)
                .withSession(this.attachment.getSession())
                .withAuthenticity(authenticity)
                .withAuthentication(this.attachment.getAuthentication())
                .withParameter(this.attachment.getRequestParameter())
                .withBody(this.attachment.getBody());
    }
745090
private function exchangeOptions()
    {
        $builder = new NodeBuilder();

        return $builder
            ->arrayNode('exchange_options')
                ->children()
                    // base info
                    ->scalarNode('name')->isRequired()->end()
                    ->scalarNode('type')
                        ->info('Set the type of the exchange. If exchange already exist - you can skip it. Otherwise behavior is unpredictable')
                    ->end()

                    // flags
                    ->booleanNode('passive')->defaultFalse()->end()
                    ->booleanNode('durable')->defaultTrue()->end()
                    ->booleanNode('auto_delete')->defaultFalse()->end()

                    // args
                    ->arrayNode('arguments')
                        ->prototype('scalar')->end()
                        ->defaultValue(array())
                        ->normalizeKeys(false)
                    ->end()

                    // binding
                    ->arrayNode('routing_keys')
                        ->prototype('scalar')->end()
                        ->defaultValue(array())
                    ->end()

                    // default message attributes
                    ->arrayNode('publish_attributes')
                        ->prototype('scalar')->end()
                        ->defaultValue(array())
                    ->end()
                ->end();
            //last end is missed here intentionally because arrayNode doesn't have an actual parent
    }
995875
public function publish(
        $exchangeName,
        $messageBody,
        $routingKey = '',
        $flags = AMQP_NOPARAM,
        array $attributes = []
    ) {
        $exchange = $this->produceExchangeInstance($exchangeName);

        $return = $exchange->publish(
            $messageBody,
            $routingKey,
            $flags,
            $attributes
        );

        return $return;
    }
594750
protected function bindQueues($queueName, $exchangeName, $routingKey = '')
    {
        $this->channel->queue_bind($queueName, $exchangeName, $routingKey);
    }
846729
public function exchangeDelete($exchange, $options)
    {
        $args = new Writer();
        $args->writeShort($options['ticket'])
            ->writeShortStr($exchange)
            ->writeBit($options['if_unused'])
            ->writeBit($options['no_wait']);
        return $args;
    }
1094096
public function ask($routing_key, $message, $timeout = 10, $exchange = null)
    {
        return $this->amqp->ask($exchange ?: $this->exchange, $routing_key, $message, $timeout);
    }
675226
public function exchangeOffset()
    {
        if ($this->count() > 0) {
            $camelcaseStorage = [];

            foreach ($this->getArrayCopy() as $offset => $value) {
                $camelcaseStorage[ camelcase($offset) ] = $value;
            }

            $this->exchangeArray($camelcaseStorage);
        }

        return $this->getArrayCopy();
    }
994258
public function createQueue($queueName)
    {
        $channel = $this->getChannel();
        $channel->exchange_declare($this->exchange, 'direct', false, true, false);
        $channel->queue_declare($queueName, false, true, false, false);
        $channel->queue_bind($queueName, $this->exchange, $queueName);
    }
702271
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);
        });
    }