+---------+-------------------+--------------------+-------------------------------------+----------+
| Type | Source | Queue | Processor | Options |
+---------+-------------------+--------------------+-------------------------------------+----------+
| command | commandName | default (prefixed) | App\Services\CommandNameProcessor | (hidden) |
+---------+-------------------+--------------------+-------------------------------------+----------+enqueue:
default:
transport:
dsn: '%env(resolve:ENQUEUE_DSN)%'
client:
traceable_producer: true
prefix: enqueue
separator: .
app_name: app
router_topic: default
router_queue: default
router_processor: null
redelivered_delay_time: 0
default_queue: default
driver_options: { }
entrypoint to php and command to run a php file. I must say I'm not using enqueue cli commands because I use Phalcon framework and so I built a custom routine.The problem only occours when I call
consume method.
Hello! I'm using php-enqueue SimpleClient with the amqp.
And I have "worker" proccess that is binding to two commands.
And then I have a "producer" proccess that recieves requests via WebSocket from the client and sends command to the worker.
It required a bit of hacking to make it work. One of the main hacks is that SimpleClient uses Symphony event loop for async, and Ratchet WebSocket uses React for that. I haven't found a way to make them work together asynchronously.
So, my "producer" is not sending heartbeat to the amqp server and gets disconnected sometimes. That's why I'm trying to send a command and if it fails - I'm reinitializing SimpleClient and sending command again.
During my first try, I'm able to capture the "Library error: a socket error occurred" exception.
Then, ratchet catches an error: "The enqueue.processor property must not be set."
Not sure why it's happening, but I'll investigate further. Now that I've written it, I have a couple of ideas about how to aproach this problem :)
I'm mostly curious: is there any way to use both React and Symphony async event loops? I can't find any WebSocket implementation with Symphony async loop.
Thanks in advance for any thoughts on this!
Let me know if sharing my code will make sense.
Forgot to mention, I'm using custom processors because default ones failed to work for me for some reason I don't remember.
And to make "producer" aware of the commands, I'm using this dummy code: $this->client->bindCommand('parse', function () {}, 'ParseCallbackProcessor');. This code also gets executed during the re-initialization. Probably it has something to do with this...
Oh well, I solved my problem already :)
It tirned out that my messages gets passed by refernce to sendCommand() and during the first (unsuccesful) try it gets modified. So when I try to send my message again, after reinitializing, I'm sending already "prepared" message, which is bad, aparently.
In case if anybody is curious - here's my new code:
/**
* @param string $command
* @param string|array|JsonSerializable|Message $message
* @param bool $needReply
* @return Promise|null
*/
public function sendCommand(string $command, $message, bool $needReply = false): ?Promise
{
try {
$m1 = clone($message);
$result = $this->client->sendCommand($command, $m1, $needReply);
} catch (Exception $exception) {
$this->initClient();
$m2 = clone($message);
$result = $this->client->sendCommand($command, $m2, $needReply);
}
return $result;
}
Hi guys,
My case is that we already have Kafka running with some producers (in Java). Message payload is just json object. I was trying to create a consumer using symfony messenger + enqueue kafka transport. And then I realised that implementation is tight to the message interface that should contain body, properties, headers.
Is there a way to overwrite this logic, so I can just have plain message data like json object which I then decode to array?
enqueue:
default:
transport: "amqp://guest:guest@rabbitmq:5672"
Hi @uro , thanks for PR. Found a little bug inside AsyncCommandExtension::load() method. Instead of
$container->register($id, RunCommandProcessor::class)
->addArgument('%kernel.project_dir%', $client['timeout'])
->addTag('enqueue.processor', [
'client' => $client['name'],
'command' => $client['command_name'] ?? Commands::RUN_COMMAND,
'queue' => $client['queue_name'] ?? Commands::RUN_COMMAND,
'prefix_queue' => false,
'exclusive' => true,
])
->addTag('enqueue.transport.processor')
;it should be
$container->register($id, RunCommandProcessor::class)
->addArgument('%kernel.project_dir%')
->addArgument($client['timeout'])
->addTag('enqueue.processor', [
'client' => $client['name'],
'command' => $client['command_name'] ?? Commands::RUN_COMMAND,
'queue' => $client['queue_name'] ?? Commands::RUN_COMMAND,
'prefix_queue' => false,
'exclusive' => true,
])
->addTag('enqueue.transport.processor')
;
@makasim
Hi!
As for newest version of symfony bundle, which is a common way to send a Command to a different queue/topic (not same as from config)? Trying to customize Message object by setting
$message->setProperty(Config::TOPIC, 'any_topic');
//and than
$producer->sendCommand('a_processor_name', $message);but seems like its not working for some reason
Hi there, quick and simple question:
I am using enqueue-redis and have a very simple Processor. How to I get enqueue to redeliver a Message for a later date, i.e., Message comes to processor, something temporarily did not work, let's process this message later.
What I tried:
- Processeor::REQUEUE, but this puts a copy of the message into the queue and instantly tries to re-process it again.
- Processor:REJECT, simply discards the message, never retries it
- Throwing an exception in the processor, this also does not achieve the intended effect.
thaks in advance
