Creating a sending service for an Event Pub/Sub using Nameko and Python

googlegroup
1

I have a service that uses the rabbitmq broker I have set up. With the code below

from nameko.events import EventDispatcher, event_handler

    from nameko.rpc import rpc

    from nameko.runners import ServiceRunner

    from nameko.testing.utils import get_container

    import time

    import os

    import sys

    import eventlet

    eventlet.monkey_patch()

    AMQP_URI = 'pyamqp://guest:guest@broker'

    config = {

        'AMQP_URI': AMQP_URI  # e.g. "pyamqp://guest:guest@localhost"

    }

    class ServiceB:

        name = 'event_listen'

        @event_handler("event_dispatch", "reading_queued")

        def handle_Event(self, payload):

            print("service b received this reading: ", str(payload).upper())

    runner = ServiceRunner(config)

    runner.add_service(ServiceB)

    container_b = get_container(runner, ServiceB)

    if __name__ == '__main__':

        try:

            print('Waiting for messages')

            while True:

                runner.start()

                time.sleep(5)

        except KeyboardInterrupt:

            print('Close Out program')

            try:

                runner.stop()

                sys.exit(0)

            except SystemExit:

                os._exit(0)

And this works as far as I can tell. I am able to send messages to this service via a nameko shell. I am wondering however how would I make a python script that lets you send messages to this “receiving” service. I have scoured the internet trying to find resources on how to accomplish this without relying on the nameko shell. There does not seem to be any good resources out there. Can someone point me in the right direction?

This is something that I have tried. I have tried calling the service directly, tried using the EventDispatcher class directly, setting up a runner for this, and nothing.

from nameko.events import EventDispatcher, event_handler

from nameko.rpc import rpc

from nameko.runners import ServiceRunner

from nameko.testing.utils import get_container

from nameko.standalone.events import event_dispatcher

from nameko.standalone.rpc import ClusterRpcProxy

import time

import eventlet

eventlet.monkey_patch()

AMQP_URI = 'pyamqp://guest:guest@broker'

config = {

    'AMQP_URI': AMQP_URI  # e.g. "pyamqp://guest:guest@localhost"

}


class ServiceA:

    name = 'event_dispatch'

    dispatch = EventDispatcher()

    @rpc

    def dispatching_method(self, payload):

        print("Running Service A")

        self.dispatch("reading_queued", payload)

# runner = ServiceRunner(config)

# runner.add_service(ServiceA)

# container_a = get_container(runner, ServiceA)

print('Starting service A')

# while True:

#     runner.start()

#     time.sleep(5)

#a = ServiceA()

#ServiceA().dispatching_method("Something here")

EventDispatcher("reading_queued", 'Make me something')
2

I did figure out a way to do what I needed.

from nameko.standalone.events import event_dispatcher

import eventlet

eventlet.monkey_patch()

AMQP_URI = 'pyamqp://guest:guest@broker'

config = {

    'AMQP_URI': AMQP_URI  # e.g. "pyamqp://guest:guest@localhost"

}

def send_current_time(time):

    event_dispatcher(config)('event_dispatch', 'reading_queued', f'The current time is: {time}')

I wish the docs were a little clearer for putting together stand-alone services. Luckily I came across this thread that gave me the answer: https://discourse.nameko.io/t/dependencyprovider-that-directly-publishes-events/497