ServiceRpcProxy/ClusterRpcProxy equivalent for @event_handler

googlegroup
1

Hey,

I have a somewhat extensive project that manages/runs plugins and uses a
priority queue to route messages/objects between them. Now I am starting to
transition some of the plugins to nameko services, but I can't move
everything at once/want to leave some parts as is. I was looking for ways
to integrate nameko into this environment. I already added calling a
service (ServiceRpcProxy/ClusterRpcProxy) and dispatching events (via a
dedicated nameko-service that dispatches it on their behalf), but I am
stuck at receiving events.

Basically, I am looking for a standalone version of @event_handler where I
can specify a function to be called on certain events and don't need to run
it in a container.
I would like to avoid polling a service if at all possible.

Or, if that is not possible, how would I go about running my own
threads/processes inside a container?

Any advice would be greatly appreciated!

Best regards
Flo

2

For *dispatching* events, you can use the standalone event dispatcher found
in nameko.standalone.events (as opposed to delegating to a service to
dispatch them for you).

For *handling* events, there's nothing standalone in the nameko codebase.
However, receiving an event is little more than consuming an AMQP message.
Each handler has its own queue and event messages are routed to them based
on:

Exchange: "<service-name>.events"
Routing Key: "<event-type>"
Queue:
"evt-<dispatcher-service-name>.<event-type>--<handler-service-name>.<handler-method-name>"

To illustrate this with an example, if you had the following service
dispatching events:

class Foo:
    name = "foo"
    
    dispatch = EventDispatcher()

    def dispatch_event(self):
        self.dispatch("foo-event", "payload")

And this one handling them:

class Bar:
    name = "bar"

    @event_handler("foo", "foo-event")
    def handle_foo_event(self, payload):
        pass

Then nameko would create the following AMQP entities:

Exchange: "foo.events"
Routing Key: "foo-event"
Queue: "evt-foo.foo-event--bar.handle_foo_event"

You can emulate the Bar service completely by creating these entities
yourself and then consuming the messages from the queue. It is pretty
straightforward to do this with kombu. The payload is just JSON-encoded
into the body unless you're using a serializer other than the default.

For your other question, it is possible to start threads inside nameko.
Extensions can call ServiceContainer.spawn_managed_thread to do so. See
nameko-sentry
<https://github.com/mattbennett/nameko-sentry/blob/master/nameko_sentry.py#L31&gt; for
an example. Note that spawn_managed_thread currently gives you back the
greenthread object, but that is likely to change.

Hope that helps,
Matt.

···

On Tuesday, 16 August 2016 02:38:46 UTC+8, jung...@gmail.com wrote:

Hey,

I have a somewhat extensive project that manages/runs plugins and uses a
priority queue to route messages/objects between them. Now I am starting to
transition some of the plugins to nameko services, but I can't move
everything at once/want to leave some parts as is. I was looking for ways
to integrate nameko into this environment. I already added calling a
service (ServiceRpcProxy/ClusterRpcProxy) and dispatching events (via a
dedicated nameko-service that dispatches it on their behalf), but I am
stuck at receiving events.

Basically, I am looking for a standalone version of @event_handler where I
can specify a function to be called on certain events and don't need to run
it in a container.
I would like to avoid polling a service if at all possible.

Or, if that is not possible, how would I go about running my own
threads/processes inside a container?

Any advice would be greatly appreciated!

Best regards
Flo

3

For *dispatching* events, you can use the standalone event dispatcher
found in nameko.standalone.events (as opposed to delegating to a service to
dispatch them for you).

Ah, I somehow must have missed that.

For *handling* events, there's nothing standalone in the nameko codebase.

However, receiving an event is little more than consuming an AMQP message.
Each handler has its own queue and event messages are routed to them based
on:

Exchange: "<service-name>.events"
Routing Key: "<event-type>"
Queue:
"evt-<dispatcher-service-name>.<event-type>--<handler-service-name>.<handler-method-name>"

To illustrate this with an example, if you had the following service
dispatching events:

class Foo:
    name = "foo"
    
    dispatch = EventDispatcher()

    def dispatch_event(self):
        self.dispatch("foo-event", "payload")

And this one handling them:

class Bar:
    name = "bar"

    @event_handler("foo", "foo-event")
    def handle_foo_event(self, payload):
        pass

Then nameko would create the following AMQP entities:

Exchange: "foo.events"
Routing Key: "foo-event"
Queue: "evt-foo.foo-event--bar.handle_foo_event"

You can emulate the Bar service completely by creating these entities
yourself and then consuming the messages from the queue. It is pretty
straightforward to do this with kombu. The payload is just JSON-encoded
into the body unless you're using a serializer other than the default.

Great! Got both your suggestions already working. Exactly what I wanted!

For your other question, it is possible to start threads inside nameko.
Extensions can call ServiceContainer.spawn_managed_thread to do so. See
nameko-sentry
<https://github.com/mattbennett/nameko-sentry/blob/master/nameko_sentry.py#L31&gt; for
an example. Note that spawn_managed_thread currently gives you back the
greenthread object, but that is likely to change.

I'm assuming any further thread spawning (from that one) should also use

ServiceContainer.spawn_managed_thread?

Hope that helps,
Matt.

Yes, thank you. That was precisely the information I was looking for!!

best regards,
Flo

···

Am Dienstag, 16. August 2016 07:46:08 UTC+2 schrieb Matt Bennett:

On Tuesday, 16 August 2016 02:38:46 UTC+8, jung...@gmail.com wrote:

Hey,

I have a somewhat extensive project that manages/runs plugins and uses a
priority queue to route messages/objects between them. Now I am starting to
transition some of the plugins to nameko services, but I can't move
everything at once/want to leave some parts as is. I was looking for ways
to integrate nameko into this environment. I already added calling a
service (ServiceRpcProxy/ClusterRpcProxy) and dispatching events (via a
dedicated nameko-service that dispatches it on their behalf), but I am
stuck at receiving events.

Basically, I am looking for a standalone version of @event_handler where
I can specify a function to be called on certain events and don't need to
run it in a container.
I would like to avoid polling a service if at all possible.

Or, if that is not possible, how would I go about running my own
threads/processes inside a container?

Any advice would be greatly appreciated!

Best regards
Flo

4

For *dispatching* events, you can use the standalone event dispatcher
found in nameko.standalone.events (as opposed to delegating to a service to
dispatch them for you).

Ah, I somehow must have missed that.

For *handling* events, there's nothing standalone in the nameko codebase.

However, receiving an event is little more than consuming an AMQP message.
Each handler has its own queue and event messages are routed to them based
on:

Exchange: "<service-name>.events"
Routing Key: "<event-type>"
Queue:
"evt-<dispatcher-service-name>.<event-type>--<handler-service-name>.<handler-method-name>"

To illustrate this with an example, if you had the following service
dispatching events:

class Foo:
    name = "foo"
    
    dispatch = EventDispatcher()

    def dispatch_event(self):
        self.dispatch("foo-event", "payload")

And this one handling them:

class Bar:
    name = "bar"

    @event_handler("foo", "foo-event")
    def handle_foo_event(self, payload):
        pass

Then nameko would create the following AMQP entities:

Exchange: "foo.events"
Routing Key: "foo-event"
Queue: "evt-foo.foo-event--bar.handle_foo_event"

You can emulate the Bar service completely by creating these entities
yourself and then consuming the messages from the queue. It is pretty
straightforward to do this with kombu. The payload is just JSON-encoded
into the body unless you're using a serializer other than the default.

Great! Got both your suggestions already working. Exactly what I wanted!

Awesome

For your other question, it is possible to start threads inside nameko.

Extensions can call ServiceContainer.spawn_managed_thread to do so. See
nameko-sentry
<https://github.com/mattbennett/nameko-sentry/blob/master/nameko_sentry.py#L31&gt; for
an example. Note that spawn_managed_thread currently gives you back the
greenthread object, but that is likely to change.

I'm assuming any further thread spawning (from that one) should also use

ServiceContainer.spawn_managed_thread?

Actually, no. The spawned thread just runs a single function or method
until it exits. It is intended to allow Extensions to run background
threads if they need to, and is otherwise quite limited.

Hope that helps,

···

On Tuesday, August 16, 2016 at 8:34:35 PM UTC+8, jung...@gmail.com wrote:

Am Dienstag, 16. August 2016 07:46:08 UTC+2 schrieb Matt Bennett:

Matt.

Yes, thank you. That was precisely the information I was looking for!!

best regards,
Flo

On Tuesday, 16 August 2016 02:38:46 UTC+8, jung...@gmail.com wrote:

Hey,

I have a somewhat extensive project that manages/runs plugins and uses a
priority queue to route messages/objects between them. Now I am starting to
transition some of the plugins to nameko services, but I can't move
everything at once/want to leave some parts as is. I was looking for ways
to integrate nameko into this environment. I already added calling a
service (ServiceRpcProxy/ClusterRpcProxy) and dispatching events (via a
dedicated nameko-service that dispatches it on their behalf), but I am
stuck at receiving events.

Basically, I am looking for a standalone version of @event_handler where
I can specify a function to be called on certain events and don't need to
run it in a container.
I would like to avoid polling a service if at all possible.

Or, if that is not possible, how would I go about running my own
threads/processes inside a container?

Any advice would be greatly appreciated!

Best regards
Flo

5

I misread your question slightly. Technically a spawned thread *can* spawn
another one, as long as it has access to the container. But it would be
unusual and outside the expected usage of that API.

···

On Tuesday, August 16, 2016 at 10:24:12 PM UTC+8, Matt Bennett wrote:

On Tuesday, August 16, 2016 at 8:34:35 PM UTC+8, jung...@gmail.com wrote:

Am Dienstag, 16. August 2016 07:46:08 UTC+2 schrieb Matt Bennett:

For *dispatching* events, you can use the standalone event dispatcher
found in nameko.standalone.events (as opposed to delegating to a service to
dispatch them for you).

Ah, I somehow must have missed that.

For *handling* events, there's nothing standalone in the nameko

codebase. However, receiving an event is little more than consuming an AMQP
message. Each handler has its own queue and event messages are routed to
them based on:

Exchange: "<service-name>.events"
Routing Key: "<event-type>"
Queue:
"evt-<dispatcher-service-name>.<event-type>--<handler-service-name>.<handler-method-name>"

To illustrate this with an example, if you had the following service
dispatching events:

class Foo:
    name = "foo"
    
    dispatch = EventDispatcher()

    def dispatch_event(self):
        self.dispatch("foo-event", "payload")

And this one handling them:

class Bar:
    name = "bar"

    @event_handler("foo", "foo-event")
    def handle_foo_event(self, payload):
        pass

Then nameko would create the following AMQP entities:

Exchange: "foo.events"
Routing Key: "foo-event"
Queue: "evt-foo.foo-event--bar.handle_foo_event"

You can emulate the Bar service completely by creating these entities
yourself and then consuming the messages from the queue. It is pretty
straightforward to do this with kombu. The payload is just JSON-encoded
into the body unless you're using a serializer other than the default.

Great! Got both your suggestions already working. Exactly what I wanted!

Awesome

For your other question, it is possible to start threads inside nameko.

Extensions can call ServiceContainer.spawn_managed_thread to do so. See
nameko-sentry
<https://github.com/mattbennett/nameko-sentry/blob/master/nameko_sentry.py#L31&gt; for
an example. Note that spawn_managed_thread currently gives you back the
greenthread object, but that is likely to change.

I'm assuming any further thread spawning (from that one) should also use

ServiceContainer.spawn_managed_thread?

Actually, no. The spawned thread just runs a single function or method
until it exits. It is intended to allow Extensions to run background
threads if they need to, and is otherwise quite limited.

Hope that helps,

Matt.

Yes, thank you. That was precisely the information I was looking for!!

best regards,
Flo

On Tuesday, 16 August 2016 02:38:46 UTC+8, jung...@gmail.com wrote:

Hey,

I have a somewhat extensive project that manages/runs plugins and uses
a priority queue to route messages/objects between them. Now I am starting
to transition some of the plugins to nameko services, but I can't move
everything at once/want to leave some parts as is. I was looking for ways
to integrate nameko into this environment. I already added calling a
service (ServiceRpcProxy/ClusterRpcProxy) and dispatching events (via a
dedicated nameko-service that dispatches it on their behalf), but I am
stuck at receiving events.

Basically, I am looking for a standalone version of @event_handler
where I can specify a function to be called on certain events and don't
need to run it in a container.
I would like to avoid polling a service if at all possible.

Or, if that is not possible, how would I go about running my own
threads/processes inside a container?

Any advice would be greatly appreciated!

Best regards
Flo

6

So what it comes down to is that I best follow my initial plan and use
standalone versions of rpc, dispatch and event_handler?

···

Am Dienstag, 16. August 2016 16:28:20 UTC+2 schrieb Matt Yule-Bennett:

I misread your question slightly. Technically a spawned thread *can*
spawn another one, as long as it has access to the container. But it would
be unusual and outside the expected usage of that API.

On Tuesday, August 16, 2016 at 10:24:12 PM UTC+8, Matt Bennett wrote:

On Tuesday, August 16, 2016 at 8:34:35 PM UTC+8, jung...@gmail.com wrote:

Am Dienstag, 16. August 2016 07:46:08 UTC+2 schrieb Matt Bennett:

For *dispatching* events, you can use the standalone event dispatcher
found in nameko.standalone.events (as opposed to delegating to a service to
dispatch them for you).

Ah, I somehow must have missed that.

For *handling* events, there's nothing standalone in the nameko

codebase. However, receiving an event is little more than consuming an AMQP
message. Each handler has its own queue and event messages are routed to
them based on:

Exchange: "<service-name>.events"
Routing Key: "<event-type>"
Queue:
"evt-<dispatcher-service-name>.<event-type>--<handler-service-name>.<handler-method-name>"

To illustrate this with an example, if you had the following service
dispatching events:

class Foo:
    name = "foo"
    
    dispatch = EventDispatcher()

    def dispatch_event(self):
        self.dispatch("foo-event", "payload")

And this one handling them:

class Bar:
    name = "bar"

    @event_handler("foo", "foo-event")
    def handle_foo_event(self, payload):
        pass

Then nameko would create the following AMQP entities:

Exchange: "foo.events"
Routing Key: "foo-event"
Queue: "evt-foo.foo-event--bar.handle_foo_event"

You can emulate the Bar service completely by creating these entities
yourself and then consuming the messages from the queue. It is pretty
straightforward to do this with kombu. The payload is just JSON-encoded
into the body unless you're using a serializer other than the default.

Great! Got both your suggestions already working. Exactly what I wanted!

Awesome

For your other question, it is possible to start threads inside nameko.

Extensions can call ServiceContainer.spawn_managed_thread to do so.
See nameko-sentry
<https://github.com/mattbennett/nameko-sentry/blob/master/nameko_sentry.py#L31&gt; for
an example. Note that spawn_managed_thread currently gives you back
the greenthread object, but that is likely to change.

I'm assuming any further thread spawning (from that one) should also

use ServiceContainer.spawn_managed_thread?

Actually, no. The spawned thread just runs a single function or method
until it exits. It is intended to allow Extensions to run background
threads if they need to, and is otherwise quite limited.

Hope that helps,

Matt.

Yes, thank you. That was precisely the information I was looking for!!

best regards,
Flo

On Tuesday, 16 August 2016 02:38:46 UTC+8, jung...@gmail.com wrote:

Hey,

I have a somewhat extensive project that manages/runs plugins and uses
a priority queue to route messages/objects between them. Now I am starting
to transition some of the plugins to nameko services, but I can't move
everything at once/want to leave some parts as is. I was looking for ways
to integrate nameko into this environment. I already added calling a
service (ServiceRpcProxy/ClusterRpcProxy) and dispatching events (via a
dedicated nameko-service that dispatches it on their behalf), but I am
stuck at receiving events.

Basically, I am looking for a standalone version of @event_handler
where I can specify a function to be called on certain events and don't
need to run it in a container.
I would like to avoid polling a service if at all possible.

Or, if that is not possible, how would I go about running my own
threads/processes inside a container?

Any advice would be greatly appreciated!

Best regards
Flo

7

That would be my recommendation, yes.

···

On Wednesday, August 17, 2016 at 2:05:11 AM UTC+8, jung...@gmail.com wrote:

So what it comes down to is that I best follow my initial plan and use
standalone versions of rpc, dispatch and event_handler?

Am Dienstag, 16. August 2016 16:28:20 UTC+2 schrieb Matt Yule-Bennett:

I misread your question slightly. Technically a spawned thread *can*
spawn another one, as long as it has access to the container. But it would
be unusual and outside the expected usage of that API.

On Tuesday, August 16, 2016 at 10:24:12 PM UTC+8, Matt Bennett wrote:

On Tuesday, August 16, 2016 at 8:34:35 PM UTC+8, jung...@gmail.com >>> wrote:

Am Dienstag, 16. August 2016 07:46:08 UTC+2 schrieb Matt Bennett:

For *dispatching* events, you can use the standalone event dispatcher
found in nameko.standalone.events (as opposed to delegating to a service to
dispatch them for you).

Ah, I somehow must have missed that.

For *handling* events, there's nothing standalone in the nameko

codebase. However, receiving an event is little more than consuming an AMQP
message. Each handler has its own queue and event messages are routed to
them based on:

Exchange: "<service-name>.events"
Routing Key: "<event-type>"
Queue:
"evt-<dispatcher-service-name>.<event-type>--<handler-service-name>.<handler-method-name>"

To illustrate this with an example, if you had the following service
dispatching events:

class Foo:
    name = "foo"
    
    dispatch = EventDispatcher()

    def dispatch_event(self):
        self.dispatch("foo-event", "payload")

And this one handling them:

class Bar:
    name = "bar"

    @event_handler("foo", "foo-event")
    def handle_foo_event(self, payload):
        pass

Then nameko would create the following AMQP entities:

Exchange: "foo.events"
Routing Key: "foo-event"
Queue: "evt-foo.foo-event--bar.handle_foo_event"

You can emulate the Bar service completely by creating these entities
yourself and then consuming the messages from the queue. It is pretty
straightforward to do this with kombu. The payload is just JSON-encoded
into the body unless you're using a serializer other than the default.

Great! Got both your suggestions already working. Exactly what I

wanted!

Awesome

For your other question, it is possible to start threads inside nameko.

Extensions can call ServiceContainer.spawn_managed_thread to do so.
See nameko-sentry
<https://github.com/mattbennett/nameko-sentry/blob/master/nameko_sentry.py#L31&gt; for
an example. Note that spawn_managed_thread currently gives you back
the greenthread object, but that is likely to change.

I'm assuming any further thread spawning (from that one) should also

use ServiceContainer.spawn_managed_thread?

Actually, no. The spawned thread just runs a single function or method
until it exits. It is intended to allow Extensions to run background
threads if they need to, and is otherwise quite limited.

Hope that helps,

Matt.

Yes, thank you. That was precisely the information I was looking for!!

best regards,
Flo

On Tuesday, 16 August 2016 02:38:46 UTC+8, jung...@gmail.com wrote:

Hey,

I have a somewhat extensive project that manages/runs plugins and
uses a priority queue to route messages/objects between them. Now I am
starting to transition some of the plugins to nameko services, but I can't
move everything at once/want to leave some parts as is. I was looking for
ways to integrate nameko into this environment. I already added calling a
service (ServiceRpcProxy/ClusterRpcProxy) and dispatching events (via a
dedicated nameko-service that dispatches it on their behalf), but I am
stuck at receiving events.

Basically, I am looking for a standalone version of @event_handler
where I can specify a function to be called on certain events and don't
need to run it in a container.
I would like to avoid polling a service if at all possible.

Or, if that is not possible, how would I go about running my own
threads/processes inside a container?

Any advice would be greatly appreciated!

Best regards
Flo

8

Everything worked out great and has been running for a couple of months
now, so thank you for all the help!

···

Am Mittwoch, 17. August 2016 06:06:48 UTC+2 schrieb Matt Yule-Bennett:

That would be my recommendation, yes.

On Wednesday, August 17, 2016 at 2:05:11 AM UTC+8, jung...@gmail.com > wrote:

So what it comes down to is that I best follow my initial plan and use
standalone versions of rpc, dispatch and event_handler?

Am Dienstag, 16. August 2016 16:28:20 UTC+2 schrieb Matt Yule-Bennett:

I misread your question slightly. Technically a spawned thread *can*
spawn another one, as long as it has access to the container. But it would
be unusual and outside the expected usage of that API.

On Tuesday, August 16, 2016 at 10:24:12 PM UTC+8, Matt Bennett wrote:

On Tuesday, August 16, 2016 at 8:34:35 PM UTC+8, jung...@gmail.com >>>> wrote:

Am Dienstag, 16. August 2016 07:46:08 UTC+2 schrieb Matt Bennett:

For *dispatching* events, you can use the standalone event
dispatcher found in nameko.standalone.events (as opposed to delegating to a
service to dispatch them for you).

Ah, I somehow must have missed that.

For *handling* events, there's nothing standalone in the nameko

codebase. However, receiving an event is little more than consuming an AMQP
message. Each handler has its own queue and event messages are routed to
them based on:

Exchange: "<service-name>.events"
Routing Key: "<event-type>"
Queue:
"evt-<dispatcher-service-name>.<event-type>--<handler-service-name>.<handler-method-name>"

To illustrate this with an example, if you had the following service
dispatching events:

class Foo:
    name = "foo"
    
    dispatch = EventDispatcher()

    def dispatch_event(self):
        self.dispatch("foo-event", "payload")

And this one handling them:

class Bar:
    name = "bar"

    @event_handler("foo", "foo-event")
    def handle_foo_event(self, payload):
        pass

Then nameko would create the following AMQP entities:

Exchange: "foo.events"
Routing Key: "foo-event"
Queue: "evt-foo.foo-event--bar.handle_foo_event"

You can emulate the Bar service completely by creating these
entities yourself and then consuming the messages from the queue. It is
pretty straightforward to do this with kombu. The payload is just
JSON-encoded into the body unless you're using a serializer other than the
default.

Great! Got both your suggestions already working. Exactly what I

wanted!

Awesome

For your other question, it is possible to start threads inside

nameko. Extensions can call ServiceContainer.spawn_managed_thread to
do so. See nameko-sentry
<https://github.com/mattbennett/nameko-sentry/blob/master/nameko_sentry.py#L31&gt; for
an example. Note that spawn_managed_thread currently gives you back
the greenthread object, but that is likely to change.

I'm assuming any further thread spawning (from that one) should also

use ServiceContainer.spawn_managed_thread?

Actually, no. The spawned thread just runs a single function or method
until it exits. It is intended to allow Extensions to run background
threads if they need to, and is otherwise quite limited.

Hope that helps,

Matt.

Yes, thank you. That was precisely the information I was looking for!!

best regards,
Flo

On Tuesday, 16 August 2016 02:38:46 UTC+8, jung...@gmail.com wrote:

Hey,

I have a somewhat extensive project that manages/runs plugins and
uses a priority queue to route messages/objects between them. Now I am
starting to transition some of the plugins to nameko services, but I can't
move everything at once/want to leave some parts as is. I was looking for
ways to integrate nameko into this environment. I already added calling a
service (ServiceRpcProxy/ClusterRpcProxy) and dispatching events (via a
dedicated nameko-service that dispatches it on their behalf), but I am
stuck at receiving events.

Basically, I am looking for a standalone version of @event_handler
where I can specify a function to be called on certain events and don't
need to run it in a container.
I would like to avoid polling a service if at all possible.

Or, if that is not possible, how would I go about running my own
threads/processes inside a container?

Any advice would be greatly appreciated!

Best regards
Flo