neptune/core/management/commands/processing.py

import msgpack
from django.core.management.base import BaseCommand
from redis import StrictRedis

from core.lib.rules import process_rules
from core.util import logs

log = logs.get_logger("processing")


class Command(BaseCommand):
    def handle(self, *args, **options):
        r = StrictRedis(unix_socket_path="/var/run/socks/redis.sock", db=0)
        p = r.pubsub()
        p.psubscribe("messages")
        for message in p.listen():
            if message:
                if message["channel"] == b"messages":
                    data = message["data"]
                    try:
                        unpacked = msgpack.unpackb(data, raw=False)
                    except TypeError:
                        continue
                    process_rules(unpacked)