Oldsound rabbitmq bundle multiple consumer configuration

Question

I'm trying to implement RabbitMQ with https://github.com/php-amqplib/RabbitMqBundle and Symfony2 framework.

I've managed to make the thing work with 1 producer and 1 consumer but the problem is when i use multiple consumers.

This is my configuration:

old_sound_rabbit_mq:
    connections:
        default:
            host:     'localhost'
            port:     5672
            user:     'guest'
            password: 'guest'
            vhost:    '/'
            lazy:     false
            connection_timeout: 3
            read_write_timeout: 3

            # requires php-amqplib v2.4.1+ and PHP5.4+
            keepalive: false

            # requires php-amqplib v2.4.1+
            heartbeat: 0

            #requires php_sockets.dll
#            use_socket: true # default false
    producers:
        soccer_team_stat:
            connection:       default
            exchange_options: {name: 'soccer_team_stat_ex', type: direct}
            queue_options:    {name: 'soccer_team_stat_qu'}
        soccer_team_stat_form:
            connection: default
            exchange_options: {name: 'soccer_team_stat_ex', type: direct}
            queue_options:    {name: 'soccer_team_stat_form_qu'}
    consumers:
        soccer_team_stat:
            connection:       default
            exchange_options: {name: 'soccer_team_stat_ex', type: direct}
            queue_options:    {name: 'soccer_team_stat_qu'}
            callback:         myapp.soccer_team_stat.consume
        soccer_team_stat_form:
            connection:       default
            exchange_options: {name: 'soccer_team_stat_ex', type: direct}
            queue_options:    {name: 'soccer_team_stat_form_qu'}
            callback:         myapp.soccer_team_stat_form.consume

Service definitions:

  <services>

        <service class="MyApp\EtlBundle\Producers\SoccerTeamStatProducer" id="myapp.soccer_team_stat.produce">
            <argument type="service" id="old_sound_rabbit_mq.soccer_team_stat_producer"/>
        </service>

        <service class="MyApp\EtlBundle\Producers\SoccerTeamStatProducer" id="myapp.soccer_team_stat_form.produce">
            <argument type="service" id="old_sound_rabbit_mq.soccer_team_stat_producer"/>
        </service>

        <service class="MyApp\EtlBundle\Consumers\SoccerTeamStatConsumer" id="myapp.soccer_team_stat.consume">
            <argument type="service" id="service_container"/>
        </service>

        <service class="MyApp\EtlBundle\Consumers\SoccerTeamStatFormConsumer" id="myapp.soccer_team_stat_form.consume">
            <argument type="service" id="service_container"/>
        </service>

    </services>

And on php app/console rabbitmq:consumer -d soccer_team_stat_form i get:

[Symfony\Component\DependencyInjection\Exception\ServiceNotFoundException] You have requested a non-existent service "old_sound_rabbit_mq.soccer_team_stat_form_consumer".

I tried various combinations includin using multiple_consumers configuration key but with no success. What i'm i missing?

Answer 1

If neither of routing_key and binding_key are set, direct exchange will behave like fanout and send the messages to all the queues it knows so based on what I'm seeing from your configurations, you are better off using fanout so do like below.

old_sound_rabbit_mq:
    connections:
        default:
            host:     %rabbit_mq_host%
            port:     %rabbit_mq_port%
            user:     %rabbit_mq_user%
            password: %rabbit_mq_pswd%
            vhost:    /
            lazy:     true
    producers:
        soccer_team_stat:
            connection:       default
            exchange_options: { name: 'soccer_team_stat_ex', type: fanout }
        soccer_team_stat_form:
            connection:       default
            exchange_options: { name: 'soccer_team_stat_form_ex', type: fanout }
    consumers:
        soccer_team_stat:
            connection:       default
            exchange_options: { name: 'soccer_team_stat_ex', type: fanout }
            queue_options:    { name: 'soccer_team_stat_qu' }
            callback:         myapp.soccer_team_stat.consume
        soccer_team_stat_form:
            connection:       default
            exchange_options: { name: 'soccer_team_stat_form_ex', type: fanout }
            queue_options:    { name: 'soccer_team_stat_form_qu' }
            callback:         myapp.soccer_team_stat_form.consume

This RabbitMQ fanout example with symfony including 2 Producer & 2 Exchange & 2 Queue & N Worker & 2 Consumer is the full example (actually the full answer to your question/already made version of what you want to do) that shows how things are done within symfony apps. I would suggest you to follow the pattern used there. Very easy to follow and maintain. If you want more examples, just search for RabbitMQ keyword in that blog.