Messenger component for populate guide?

[kricha]

Did anyone have completer or something near complete guide to use symfony/messenger for populating process? Seems that community need to have information about enabling async persister, using custom pager provider etc...

[saalexeev]

I use that way: messenger.yaml:

framework:
  messenger:
    transports:
      elastica: '%env(MESSENGER_TRANSPORT_DSN)%?queue_name=elastica'
    routing:
        FOS\ElasticaBundle\Message\AsyncPersistPage: elastica

services.yaml:

  FOS\ElasticaBundle\Persister\AsyncPagerPersister:
      arguments:
          - '@fos_elastica.pager_persister_registry'
          - '@fos_elastica.pager_provider_registry'
          - '@Symfony\Component\Messenger\MessageBusInterface'
      tags:
          - { name: "fos_elastica.pager_persister", persisterName: "async" }

  FOS\ElasticaBundle\Message\Handler\AsyncPersistPageHandler:
      tags: [ messenger.message_handler ]

then run in-bundle populate command. It will dispatch paged messages to 'elastica' queue. Then consume this messages with messenger:consume

may be something i''ve done wrong but it works

[kricha]

@saalexeev thanks. But question is more for solid contributors. I also have working version, but for me it seems ugly :D

[oleg-andreyev]

Enable messenger:

fos_elastica:
    messenger: ~

Configure transport

framework:
    messenger:
        transports:
            async_elasticsearch: '%env(MESSENGER_TRANSPORT_DSN)%/async_elasticsearch

        routing:
            FOS\ElasticaBundle\Message\AsyncPersistPage: async_elasticsearch

Configure handler:

    FOS\ElasticaBundle\Message\Handler\AsyncPersistPageHandler:
        arguments:
            - '@fos_elastica.async_pager_persister'
        tags: [ 'messenger.message_handler' ]

[kezek]

Thanks @oleg-andreyev and @saalexeev , do you have any idea how I can avoid downtime (0 results nodes) when using aliases with the above configuration ?

[oleg-andreyev]

This configuration does not work with "creating" index and switching aliases. It uses existing index (or creates if missing)

[kezek]

@oleg-andreyev thanks , I assumed so. The issue is that PopulateListener is triggered after the worker's dispatch. I can avoid that that appending --no-reset , but how would I know when the async workers are done in order to make the alias switch ? Any hint is appreciated.

[oleg-andreyev]

Unfortunately you cannot, it's async and order of execution is not guaranteed. The only way I can think of is monitoring queue size, and switch after it has 0 messages

[boroth]

I'm willing to write up and submit a PR for some additional documentation on how to get symfony/messenger working, but I'm currently getting "The receiver 'elasticsearch' does not exist" when trying to run the consumer with bin/console messenger/consume elasticsearch -vv.

As far as I can tell, I have things configured correctly:

app/config/services/fos_elastica.yml:

fos_elastica:
  messenger: ~
  ...

services:
  FOS\ElasticaBundle\Message\Handler\AsyncPersistPageHandler:
    arguments:
      - '@fos_elastica.async_pager_persister'
    tags: [ messenger.message_handler ]

app/config/packages/messenger.yml:

framework:
    messenger:
        transports:
            # https://symfony.com/doc/current/messenger.html#transport-configuration
            elasticsearch: "doctrine://default?queue_name=elasticsearch"

        routing:
            # Route your messages to the transports
            'FOS\ElasticaBundle\Message\AsyncPersistPage': elasticsearch

Any idea why symfony doesn't seem to see my "elasticsearch" consumer/receiver/transport? Using Symfony 4 with FOSElasticaBundle v6.

[vladdnepr]

Thanks @oleg-andreyev and @saalexeev , do you have any idea how I can avoid downtime (0 results nodes) when using aliases with the above configuration ?

Hi. I found a solution

https://gist.github.com/vladdnepr/728e1b7b428923443f25d06984034f7a

Hope this helps

[inarli]

Hey, thanks for this guide, I used it. I have another issue regarding this using style; I use messenger for populate command but, an entity is updating without using messenger system when is update on the database(orm).

✅ The populate command is processed through messenger. it is fine. ❌ Updates to the database are processing outside the queue system.

my fos_elastica.yaml like that.

fos_elastica:
    messenger:
        enabled: true

            persistence:
                driver: orm
                model:  App\Entity\User
                identifier: id
                listener:
                    defer: true
                provider: ~
                finder: ~
                elastica_to_model_transformer:
                    ignore_missing: true

What am I missing?

[boroth]

@inarli , as far as I know, the messenger is only used for the populate command, it doesn't affect the normal persister when just updating entities through the ORM.

[inarli]

yes, it seems like as you say.

@inarli , as far as I know, the messenger is only used for the populate command, it doesn't affect the normal persister when just updating entities through the ORM.

yes, it seems like as you say.