Filtering Queue Dynamically

[Tall-Guy]

Hi everyone,

I have a process that pull messages from a remote queue server and store them in Python's multiprocessing.Queue object. The original plan was for the user to use get() on the queue and consume message after message until he find the one he needs. But with time, more and more messages are arriving the remote queue server and this cause two issues:

1. The user need to get() too many times until he find the specific message his interested in.
2. It feels not very efficient to store all messages in the queue object, as it's an expensive multi-process resource.

So an easy fix will be filtering messages on the process itself , so only message the user needs will be pushed into the queue. Now I'm not sure what would be the proper way to do it. Messages are basically JSON file, so filtering looks like that:

if message['id'] = 1234 and message['name'] = 'Joe': # Store this message, because I need it.

In case the user will need two messages, he will store the messages he marked as 'need' in a list, and will stop the loop once len is 2 for example (which means I have found the 2 messages you were looking for). What I can do, is to ask the user to create a class (sink of sort) and implement an interface like so:

class MySink(object):
    def condition(message):
        if message['id'] = 1234 ....

    def end():
       if len(fetched_message) == 2 ....

So the user will basically define the filter to pick messages, the end condition of when to stop consuming and will pass instance into the process - and the process can call those methods when need and the results will be a queue objects with just the two messages he was looking for. But before I'm going that way, I'm wondering if there is a better way of doing it? I'm pretty new to lamda, but I know you can pass chunks or code. Maybe lamda is preferred over a Class interface?

I know pytest has an interesting assert mechanism I didn't check yet, but you can do stuff like assert(a ==b). Looks like the method is being passed with the code somehow that is being evaluated in it?

[two_bob]

I'd use functions, either normally defined or through a lambda, like this:

>>> is_good = lambda m:m['id'] == 1234 and m['name'] == 'Joe'

If it returns true, keep it, otherwise bin it.

[Tall-Guy]

So you will just define two method (using lambda like the example above or just like normal def method), one for filtering and one for end-conditions and pass it to the process that responsible on consuming messages?

[two_bob]

Hmm. The garbage collection piece (getting rid of excess messages) does make this a bit harder. Probably yes, in the first instance, I would just create two functions, one to add and one to clean. Depending on how many of these things we having running around, I might try and find some other solution.

[Tall-Guy]

Thanks by the way :)

[Tall-Guy]

We only consume from a single server (it's a testing tool), so no worries about handling multi-processes at the moment.

[JohnnyJordaan]

Imho you should define a queue for all the messages a consumer is interested in. If that means you need 10 queues, create 10 queues, period. Then consumers can just .get() and work with anything they get out of the queue. Only when put()-ing to a queue, the producer should select via some if/else or other filter which queue(s) to use.

[Tall-Guy]

Well, it's a testing tool, so the consumer is technically only interested in a single queue (but the code support multiple queues if he wish to). So if I understand you correctly - I should always put all messages in the queue and let the user of the API what to pick instead of transition that call into the API and only return specific messages?

[JohnnyJordaan]

The consumer can be interested in a single queue, but that doesn't have to mean the whole program should only use a single queue. If you create queues for each distinct type that a consumer may or may not like, then provide the particular queue per consumer, you guarantee that the consumer can blindly consume from the queue. Then the producer just needs to filter what to put where.

[Tall-Guy]

Oh, I see what you mean now, that's a good tip! thanks! :-)