php-enqueue/Lobby

Chris

@Kalyse

Actually, this is relevant, the consume is handled by https://php-enqueue.github.io/bundle/async_events/ it's an async processor

Chris

@Kalyse

Solved it. I ended up needing to catch the relevant Exception in my Transformer, then throwing a different type, and registeriung a extension which implements ProcessorExceptionExtensionInterface and returns a Result::Reject()

Valentin Saik

@svbackend

Hello @makasim could you please have a look to php-enqueue/enqueue-dev#1121 ? would be really cool if you can help with questions mentioned there, as I'm kind of blocked because of forma-pro docker image that doesn't support php8 yet (https://github.com/formapro/docker-nginx-php-fpm), I'm not sure you're still working there so should we rely on this docker image at all? Or would be better to move docker images to your (or enqueue) repository?

Adnan Siddiq

@adnansiddiq

Hello, Can I use php-enqueue/laravel-queue in Laravel 5.2 with php 7.0

Andrew

@andrewmy

Hey @makasim, I've sent a PR php-enqueue/enqueue-dev#1126
Your travis is still waiting but has run through on my side https://travis-ci.com/github/andrewmy/enqueue-dev/builds/213201382
It appears that the only failing tests are for pubsub, I guess nothing to do with my changes 🤷‍♂️

1 reply

Max Kotliar

@makasim

travis-ci.org is known to be slow. As far as I know it is kind of legacy and all project should migrate to travis-ci.com

Unfortunately I have no time to do the migration in near future.

There is also an issue with docker hub images. Because of it's new polices it is useless for OS and breaks CI atm.

That should be addressed too.

Without reliable CI I cannot merge PRs.

Andrew

@andrewmy

Ouch. Heard about the travis crap, probably best to move to gh actions

Max Kotliar

@makasim

Good idea. Though It's hell of a lot work.

1 reply

Andrew

@andrewmy

There's a bug in php-rdkafka, reported in their gitter — the message headers are deserialized to all strings when they can also be ints and nulls. E.g.:

RdKafka\Message {#485
  +err: 0
  +topic_name: "messages"
  +timestamp: 1610657820673
  +partition: 0
  +payload: "{"body":"some_body","properties":[],"headers":{"Content-Type":"application\/json","name":"some_name","id":"8d2977e3-695f-45e9-a09f-f1541db273da","timestamp":1610657820,"retry_count":null}}"
  +len: 188
  +key: null
  +offset: 2
  +headers: array:5 [
    "Content-Type" => "application/json"
    "name" => "some_name"
    "id" => "8d2977e3-695f-45e9-a09f-f1541db273da"
    "timestamp" => "1610657820"
    "retry_count" => ""
  ]
  errstr: "Success"
}

See timestamp and retry_count.

Now the enqueue consumer prefers those headers to its own deserialization result. Would it make sense to drop the header merge? Or reverse the preference?

Alex

@atrigub

hi guys, does somebody know how implement batch consumer?

Ivan Bragin

@aywan

Hi everyone. I need sync DB transaction with sending messages to queue or in other word send messages to RMQ after committing DB transaction. I guess, that i need to use some kind of unit-of-work pattern. Can somebody suggest ready-made solution?

Niv Penso

@nivpenso

Hi Ivan, the task you are trying to accomplish is not very clear (to me). can you please try to elaborate?

Ivan Bragin

@aywan

@nivpenso Hi! I found some solution - events. Problem that i have project without any underlying framework, and no event dispatcher. I'm not happy with events, but it is real solution for this. I try to find something, not event oriented. But this still same pattern. Now I need redesign old part of system for events.
As an example, current state:

begin transaction
update row
send message to RMQ
commit transaction

After some optimization, consumer receive message before committing transaction. Right pipeline of course:

begin transaction
update row
enqueue message
commit transaction
if transaction committed - send messages from queue.

Niv Penso

@nivpenso

you can't dispatch the event within the transaction, because, that operation has not been done yet. That will lead to race-conditions issues (your event reading consumers and db transaction are racing who perform its operation faster).

you must dispatch the event outside of the transaction

Niv Penso

@nivpenso

In case you receive a failure when enqueuing the event (sending it to RMQ), you can have several strategies to deal with it:

send the event to a dead-letter queue and handle it later
activate a retry mechanism.
log the issue to the disk.
rollback the DB row to the old values - not recommended, since there can be failures in the 2nd row update - and then it

Niv Penso

@nivpenso

btw there is also a way to do it with the transaction by adding a database table as "middle queue"
you can read about it here: https://microservices.io/patterns/data/database-triggers.html

Ivan Bragin

@aywan

@nivpenso Thanks!

Andrew

@andrewmy

@makasim could you please make a point release of https://github.com/php-enqueue/rdkafka to allow ext-rdkafka 5?

Max Kotliar

@makasim

@andrewmy https://github.com/php-enqueue/rdkafka/releases/tag/0.10.7

1 reply

Michael Frank

@ITspirit

Hi all,
i am using the enqueue Bundle and it worked great. Unfortunalty we´ve upgraded a project an there´s a new symfony version.
The described method in the DOC doesn´t work anymore:

// If you want a different producer than default (for example the other specified in sample above) then use
// $producer = $container->get('enqueue.client.some_other_transport.producer');

The service is now private as shown in the error message and can´t be accessed in the way above.
Any suggestions how i can get the producer?

Jens Prangenberg

@jprangenbergde

Hello together, does someone use the "JobRunner" to run unique jobs? Where do i find the SQL to create the doctrine tables?

Vladyslav

@byvlad

Hi all!

Vladyslav

@byvlad

I have an interesting task, but I don't know how to realize it in a short way. So, we have 10 games, each game has 10 machines. So game.#.# (game.10.10, game.10.9, game.1.3 etc). All jobs for one machine must be unique, so for example each machine like game.3.5 must have own queue. All machines need to work in parallel, but jobs in machine - one by one. All jobs for machine are exactly the same, so we can use one processor for it. But how to realize it with enqueue, I don't know. Maybe somebody have any ideas?

Muhammad Nuzaihan Bin Kamal Luddin

@muhammadn

Hi. I ran php bin/console enqueue:consume --setup-broker -vvv but the queue processes only after i do "CTRL+C" on it.. then it gets processed

I don't know why php bin/console enqueue:consume --setup-broker -vvv does not process but i had to interrupt then it starts processing

Muhammad Nuzaihan Bin Kamal Luddin

@muhammadn

Got it

it seems i cannot just do $this->mailer->send($message); for sending emails with Swift Mailer

i have to do

                    //Send the message
                    $transport = $this->containerInterface->get('swiftmailer.transport.real');
                    if (!$transport->isStarted()) {
                        $transport->start();
                    }

                    $transport->send($message);

works perfectly now

Jan Vansteenlandt

@coreation

Hello everyone! I was looking into the enqueue packages and was wondering how to use the laravel wrapper and make the configuration work for Kafka, the example that is provided (https://github.com/php-enqueue/enqueue-dev/blob/master/docs/laravel/queues.md) is using rabbitmq. I've seen a similar question here before in the history, but the only reply to that one was the OP itself with: "solved!"... :)

Happy to update some documentation pages on Github if this can be figured out

Jan Vansteenlandt

@coreation

Never mind, I worked my way through some code, I'll be making a PR for the Laravel Enqueue > Kafka. Great work on the maintainers side, btw, nicely done!

jonatansantana

@jonatansantana

Have support for Azure Service Bus?

George

@gdsmith

Hey, I’ve just added a PR to expose the sqs specific consumer methods on the snsqs consumer, would be great if someone could have a look for me php-enqueue/enqueue-dev#1160

George

@gdsmith

Thanks for the merge on that PR, how does that then make it into a release?

Max Kotliar

@makasim

Released 0.10.11 https://github.com/php-enqueue/enqueue-dev/blob/master/CHANGELOG.md#01011-2021-04-28

1 reply

Arif DEMİR

@epicentre_gitlab

Hi, how can I set the error_cb setting in the rdkafka transport.

I do it with the logic in the following url but it doesn't work.
https://github.com/getsentry/sentry-symfony/issues/336#issuecomment-746633997

burzum

@burzum

https://github.com/burzum/php-enqueue-examples can anyone confirm that running this (php Listener.php) is causing an exception? Uncaught LogicException: Processor was not found. processorName: "Enqueue\Consumption\CallbackProcessor60b402a965eb8"

When debugging ArrayProcessorRegistry.php I can see that it adds two processors, but the names don't match the one from the exception.

Tomaž

@tomazpu

Hi, i trying to configure enqueue with kafka + symfony. I have a simple Producer(CLI command) and Consumer(Processor). Im having issues to setup the kafka topic, can maybe one of you point me where and how to set it? i have the issue that both publish or consume from enqueue.app.default

Tomaž

@tomazpu

All I want is one topic for one queue, it might be mentioned in the doc, but I really cannot find it.

Niv Penso

@nivpenso

I remember I had an issue with it as well

// config/packages/enqueue.yaml
  default:
    transport:
      dsn: "rdkafka://"
      global:
        group.id: 'myapp'
        sasl.username: "%env(KAFKA_KEY)%"
        sasl.password: "%env(KAFKA_SECRET)%"
        sasl.mechanisms: "%env(KAFKA_MECHANISM)%"
        security.protocol: "%env(KAFKA_PROTOCOL)%"
        metadata.broker.list: "%env(KAFKA_BROKER_LIST)%"
      topic:
        auto.offset.reset: earliest
      commit_async: true
    client: ~

// config/packages/messenger.yaml
framework:
    messenger:
        # Uncomment this (and the failed transport below) to send failed messages to this transport for later handling.
        failure_transport: failed
        buses:
            messenger.bus.default:
                default_middleware: allow_no_handlers
        transports:
            kafka-ticket-messages:
                dsn: '%env(MESSENGER_KAFKA_TICKET_TRANSPORT_DSN)%'
                serializer: messenger.transport.symfony_serializer

        routing:
            # Route your messages to the transports
            Some\Event\TicketAddedEvent: kafka-ticket-messages

Niv Penso

@nivpenso

# .env file (environment variables)
KAFKA_KEY="some_key_here"
KAFKA_SECRET="some_secret_here"
KAFKA_BROKER_LIST="the_broker_list"
KAFKA_PROTOCOL="SASL_SSL"
KAFKA_MECHANISM="PLAIN"
MESSENGER_KAFKA_TICKET_TRANSPORT_DSN=enqueue://default?topic[name]=YOUR_TOPIC_NAME&queue[name]=YOUR_QUEUE_NAME

Niv Penso

@nivpenso

@tomazpu good luck if you need anything you can send me a pm

Where communities thrive

People

Repo info

Activity