<commit_before>import json
from collections import defaultdict
from queue import Queue

from gevent import Timeout, spawn
from tenacity import retry, wait_exponential

from CTFd.cache import cache
from CTFd.utils import string_types


class ServerSentEvent(object):
    def __init__(self, data, type=None, id=None):
        self.data = data
        self.type = type
        self.id = id

    def __str__(self):
        if isinstance(self.data, string_types):
            data = self.data
        else:
            data = json.dumps(self.data)
        lines = ["data:{value}".format(value=line) for line in data.splitlines()]
        if self.type:
            lines.insert(0, "event:{value}".format(value=self.type))
        if self.id:
            lines.append("id:{value}".format(value=self.id))
        return "\n".join(lines) + "\n\n"

    def to_dict(self):
        d = {"data": self.data}
        if self.type:
            d["type"] = self.type
        if self.id:
            d["id"] = self.id
        return d


class EventManager(object):
    def __init__(self):
        self.clients = {}

    def publish(self, data, type=None, channel="ctf"):
        event = ServerSentEvent(data, type=type)
        message = event.to_dict()
        for client in list(self.clients.values()):
            client[channel].put(message)
        return len(self.clients)

    def listen(self):
        pass

    def subscribe(self, channel="ctf"):
        q = defaultdict(Queue)
        self.clients[id(q)] = q
        try:
            while True:
                try:
                    # Immediately yield a ping event to force Response headers to be set
                    # or else some reverse proxies will incorrectly buffer SSE
                    yield ServerSentEvent(data="", type="ping")

                    with Timeout(5):
                        message = q[channel].get()
                        yield ServerSentEvent(**message)
                except Timeout:
                    yield ServerSentEvent(data="", type="ping")
        finally:
            del self.clients[id(q)]
            del q


class RedisEventManager(EventManager):
    def __init__(self):
        super(EventManager, self).__init__()
        self.client = cache.cache._write_client
        self.clients = {}

    def publish(self, data, type=None, channel="ctf"):
        event = ServerSentEvent(data, type=type)
        message = json.dumps(event.to_dict())
        return self.client.publish(message=message, channel=channel)

    def listen(self, channel="ctf"):
        @retry(wait=wait_exponential(min=1, max=30))
        def _listen():
            while True:
                pubsub = self.client.pubsub()
                pubsub.subscribe(channel)
                try:
                    while True:
                        message = pubsub.get_message(
                            ignore_subscribe_messages=True, timeout=5
                        )
                        if message:
                            if message["type"] == "message":
                                event = json.loads(message["data"])
                                for client in list(self.clients.values()):
                                    client[channel].put(event)
                finally:
                    pubsub.close()

        spawn(_listen)

    def subscribe(self, channel="ctf"):
        q = defaultdict(Queue)
        self.clients[id(q)] = q
        try:
            while True:
                try:
                    # Immediately yield a ping event to force Response headers to be set
                    # or else some reverse proxies will incorrectly buffer SSE
                    yield ServerSentEvent(data="", type="ping")

                    with Timeout(5):
                        message = q[channel].get()
                        yield ServerSentEvent(**message)<commit_msg>reduce ping events (Fix #1685) (#1699)<commit_after>import json
from collections import defaultdict
from queue import Queue

from gevent import Timeout, spawn
from tenacity import retry, wait_exponential

from CTFd.cache import cache
from CTFd.utils import string_types


class ServerSentEvent(object):
    def __init__(self, data, type=None, id=None):
        self.data = data
        self.type = type
        self.id = id

    def __str__(self):
        if isinstance(self.data, string_types):
            data = self.data
        else:
            data = json.dumps(self.data)
        lines = ["data:{value}".format(value=line) for line in data.splitlines()]
        if self.type:
            lines.insert(0, "event:{value}".format(value=self.type))
        if self.id:
            lines.append("id:{value}".format(value=self.id))
        return "\n".join(lines) + "\n\n"

    def to_dict(self):
        d = {"data": self.data}
        if self.type:
            d["type"] = self.type
        if self.id:
            d["id"] = self.id
        return d


class EventManager(object):
    def __init__(self):
        self.clients = {}

    def publish(self, data, type=None, channel="ctf"):
        event = ServerSentEvent(data, type=type)
        message = event.to_dict()
        for client in list(self.clients.values()):
            client[channel].put(message)
        return len(self.clients)

    def listen(self):
        pass

    def subscribe(self, channel="ctf"):
        q = defaultdict(Queue)
        self.clients[id(q)] = q
        try:
            # Immediately yield a ping event to force Response headers to be set
            # or else some reverse proxies will incorrectly buffer SSE
            yield ServerSentEvent(data="", type="ping")
            while True:
                try:
                    with Timeout(5):
                        message = q[channel].get()
                        yield ServerSentEvent(**message)
                except Timeout:
                    yield ServerSentEvent(data="", type="ping")
        finally:
            del self.clients[id(q)]
            del q


class RedisEventManager(EventManager):
    def __init__(self):
        super(EventManager, self).__init__()
        self.client = cache.cache._write_client
        self.clients = {}

    def publish(self, data, type=None, channel="ctf"):
        event = ServerSentEvent(data, type=type)
        message = json.dumps(event.to_dict())
        return self.client.publish(message=message, channel=channel)

    def listen(self, channel="ctf"):
        @retry(wait=wait_exponential(min=1, max=30))
        def _listen():
            while True:
                pubsub = self.client.pubsub()
                pubsub.subscribe(channel)
                try:
                    while True:
                        message = pubsub.get_message(
                            ignore_subscribe_messages=True, timeout=5
                        )
                        if message:
                            if message["type"] == "message":
                                event = json.loads(message["data"])
                                for client in list(self.clients.values()):
                                    client[channel].put(event)
                finally:
                    pubsub.close()

        spawn(_listen)

    def subscribe(self, channel="ctf"):
        q = defaultdict(Queue)
        self.clients[id(q)] = q
        try:
            # Immediately yield a ping event to force Response headers to be set
            # or else some reverse proxies will incorrectly buffer SSE
            yield ServerSentEvent(data="", type="ping")
            while True:
                try:
                    with Timeout(5):
                        message = q[channel].get()
                        yield ServerSentEvent(**message)
{'fraction_non_alphanumeric': 0.1027831715210356, 'fraction_numerical': 0.0028478964401294497, 'mean_word_length': 1.63506139154161, 'pattern_counts': {'":': 4, '<': 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': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '899063', 'n_tokens_mistral': 2099, 'n_tokens_neox': 2006, 'n_words': 514}