<commit_before>import asyncio

from copy import copy
from time import sleep

from logbook import Logger
from queue import Queue

from riemann_client.client import QueuedClient
from riemann_client.transport import RiemannError


logger = Logger("Processor")


class StopEvent(object):
    pass


class AugmentFixture(object):

    def apply_augment(self, data):
        if "service" not in data or hasattr(self, "stopped"):
            return

        key = data["service"]
        if key not in self.augments:
            return

        subscribers = self.augments[key]
        for sub in subscribers:
            sub.put_nowait(data)

    def on_stop(self):
        if hasattr(self, "stopped"):
            return

        logger.info("Stopping all augments")

        self.stopped = True
        for _, subscribers in self.augments.items():
            for sub in subscribers:
                sub.put_nowait(StopEvent())

        for _, subscribers in self.augments.items():
            for sub in subscribers:
                while not sub.empty():
                    sleep(0.1)


class QClient(QueuedClient, AugmentFixture):
    def __init__(self, *args, **kwargs):
        self.augments = {}
        self.tasks = []

    def event(self, **data):
        self.tasks.append(self.apply_augment(copy(data)))
        super(QClient, self).event(**data)

    def flush(self, transport):
        response = self.transport.send(self.queue)
        self.clear_queue()<commit_msg>Implement queue reset<commit_after>import asyncio

from copy import copy
from time import sleep

from logbook import Logger
from queue import Queue

from riemann_client.client import QueuedClient
from riemann_client.transport import RiemannError
from riemann_client import riemann_pb2


logger = Logger("Processor")


class StopEvent(object):
    pass


class AugmentFixture(object):

    def apply_augment(self, data):
        if "service" not in data or hasattr(self, "stopped"):
            return

        key = data["service"]
        if key not in self.augments:
            return

        subscribers = self.augments[key]
        for sub in subscribers:
            sub.put_nowait(data)

    def on_stop(self):
        if hasattr(self, "stopped"):
            return

        logger.info("Stopping all augments")

        self.stopped = True
        for _, subscribers in self.augments.items():
            for sub in subscribers:
                sub.put_nowait(StopEvent())

        for _, subscribers in self.augments.items():
            for sub in subscribers:
                while not sub.empty():
                    sleep(0.1)


class QClient(QueuedClient, AugmentFixture):
    def __init__(self, *args, **kwargs):
        self.augments = {}
        self.tasks = []
        self.queue = riemann_pb2.Msg()

    def event(self, **data):
        self.tasks.append(self.apply_augment(copy(data)))
        super(QClient, self).event(**data)

    def clear_queue(self):
        self.queue = riemann_pb2.Msg()

    def flush(self, transport):
        response = self.transport.send(self.queue)
        self.clear_queue()
{'fraction_non_alphanumeric': 0.09500805152979067, 'fraction_numerical': 0.0022544283413848632, 'mean_word_length': 2.3542116630669545, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6477797', 'n_tokens_mistral': 927, 'n_tokens_neox': 864, 'n_words': 237}