0
\$\begingroup\$

I have a stream of ZMQ messages, I need to filter the right ones and save them in MongoDB. The trick is to have an option to set the filter conditions dynamically.

I have come up with this solution using multiprocessing and Flask. Basicaly I use Flask HTTP API to change the global filter_parameters dictionary given by Manager and shared with another Process that is running the zmq filter itself. So my question is - is it a good solution and what is the best practice in this regard?

from multiprocessing import Process, Manager
from flask import Flask, jsonify, request
from pymongo import MongoClient

app = Flask(__name__)
manager = Manager()
filter_parameters = manager.dict()
filter_parameters['addresses'] = ['']

def scan(q, filter_parameters):
    import zmq
    context = zmq.Context()
    socket = context.socket(zmq.SUB)
    socket.connect("tcp://localhost:5556")
    socket.setsockopt_unicode(zmq.SUBSCRIBE, 'tx')
    client = MongoClient() # localhost
    db = client.transactions_db
    tx_collection = db.tx_collection
    while True:
        recv_str = socket.recv_string()
        recv_str = recv_str.split()
        filter(recv_str, filter_parameters, tx_collection)

def filter(recv_str, filter_parameters, tx_collection):
    if recv_str[0] == 'tx':
        if recv_str[7] == '0':
            if recv_str[2] in filter_parameters['addresses']:
                tx_collection.insert_one({'tx':recv_str}).inserted_id

p = Process(target=scan, args=(filter_parameters))
p.daemon = True
p.start()

@app.route('/zmq_buffer/set_filter_addresses', methods=["POST"])
def set_filter_addresses():
    data_jsn = request.json
    filter_parameters['addresses'] = data_jsn['addresses']
    return jsonify({'addresses':filter_parameters['addresses']}), 200

@app.route('/zmq_buffer/get_filter_addresses', methods=["GET"])
def get_filter_addresses():
    if filter_parameters['addresses']:
        resp = jsonify(filter_parameters['addresses'])
    else:
        resp = 'None'
    return resp, 200

if __name__ == '__main__':
    app.run(port=7050)
    p.terminate()
\$\endgroup\$
  • \$\begingroup\$ Why are you using multiprocessing? \$\endgroup\$ – Peilonrayz Mar 20 at 15:06
  • \$\begingroup\$ @Peilonrayz Because how else you would separate HTTP requests and ZMQ stream handling? \$\endgroup\$ – stkubr Mar 20 at 15:29

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.