exchangeRepository = $exchangeRepository; $this->envelopeFactory = $envelopeFactory; $this->messageEncoder = $messageEncoder; $this->messageValidator = $messageValidator; $this->responseQueueNameBuilder = $responseQueueNameBuilder; $this->publisherConfig = $publisherConfig; $this->messageIdGenerator = $messageIdGenerator; } /** * @inheritdoc */ public function publish($topicName, $data) { $envelopes = []; $replyTo = $this->responseQueueNameBuilder->getQueueName($topicName); foreach ($data as $message) { $this->messageValidator->validate($topicName, $message); $message = $this->messageEncoder->encode($topicName, $message); $envelope = $this->envelopeFactory->create( [ 'body' => $message, 'properties' => [ 'reply_to' => $replyTo, 'delivery_mode' => 2, 'correlation_id' => rand(), 'message_id' => $this->messageIdGenerator->generate($topicName), ] ] ); $envelopes[] = $envelope; } $publisher = $this->publisherConfig->getPublisher($topicName); $connectionName = $publisher->getConnection()->getName(); $exchange = $this->exchangeRepository->getByConnectionName($connectionName); return $exchange->enqueue($topicName, $envelopes); } }