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