<commit_before>##
# See the file COPYRIGHT for copyright information.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
##

"""
HTML5 EventSource support.
"""

from collections import deque
from time import time

from zope.interface import implementer
from twisted.logger import Logger, ILogObserver

from ..data.model import Incident
from ..data.json import textFromJSON



class Event(object):
    """
    HTML5 EventSource event.
    """
    def __init__(self, message, eventID=None, eventClass=None, retry=None):
        self.message    = message
        self.eventID    = eventID
        self.eventClass = eventClass
        self.retry      = retry


    def render(self):
        parts = []

        if self.eventID is not None:
            parts.append(u"id: {0}".format(self.eventID))

        if self.eventClass is not None:
            parts.append(u"event: {0}".format(self.eventClass))

        if self.retry is not None:
            parts.append(u"retry: {0:d}".format(self.retry))

        parts.extend(
            u"data: {}".format(line) for line in self.message.split(u"\n")
        )

        return (u"\r\n".join(parts) + u"\r\n\r\n")



@implementer(ILogObserver)
class DataStoreEventSourceLogObserver(object):
    """
    Observer for events related to any updates to the data store.
    """
    log = Logger()


    def __init__(self):
        self._listeners = set()
        self._events = deque(maxlen=1000)
        self._start = time()
        self._counter = 0


    def addListener(self, listener, lastEventID=None):
        self.log.debug("Adding listener: {listener}", listener=listener)

        self._playback(listener, lastEventID)

        self._listeners.add(listener)


    def removeListener(self, listener):
        self.log.debug("Removing listener: {listener}", listener=listener)

        self._listeners.add(listener)


    def _transmogrify(self, loggerEvent, eventID):
        """
        Convert a logger event into an EventSource event.
        """
        if "storeWriteClass" not in loggerEvent:
            # Not a data store event
            return None

        eventClass = loggerEvent.get("storeWriteClass", None)

        if eventClass is None:
            return

        elif eventClass is Incident:
            incident = loggerEvent.get("incident", None)

            if incident is None:
                incidentNumber = loggerEvent.get("incidentNumber", None)
            else:
                incidentNumber = incident.number

            if incidentNumber is None:
                self.log.critical(
                    "Unable to determine incident number from store event: "
                    "{event}",
                    event=loggerEvent,
                )
                return

            message = dict(incident_number=incidentNumber)

        else:
            self.log.debug(
                "Unknown data store event class: {eventClass}",
                eventClass=eventClass
            )
            return

        eventSourceEvent = Event(
            eventID=eventID,
            eventClass=eventClass.__name__,
            message=textFromJSON(message),
        )
        return eventSourceEvent


    def _playback(self, listener, lastEventID):
        if lastEventID is None:
            return

        observerID, counter = lastEventID.split(":")

        if observerID != str(id(self)):
            # lastEventID came from a different observer
            counter = 0

        for eventCounter, event in self.events:
            if eventCounter >= counter:
                listener.write(event.render().encode("utf-8"))


    def _publish(self, eventSourceEvent, eventID):
        eventText = eventSourceEvent.render().encode("utf-8")

        for listener in self._listeners:
            listener.write(eventText)

        self._events.append((self._counter, eventSourceEvent))

<commit_msg>If we can't write to a listener, drop it.<commit_after>##
# See the file COPYRIGHT for copyright information.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
##

"""
HTML5 EventSource support.
"""

from collections import deque
from time import time

from zope.interface import implementer
from twisted.logger import Logger, ILogObserver

from ..data.model import Incident
from ..data.json import textFromJSON



class Event(object):
    """
    HTML5 EventSource event.
    """
    def __init__(self, message, eventID=None, eventClass=None, retry=None):
        self.message    = message
        self.eventID    = eventID
        self.eventClass = eventClass
        self.retry      = retry


    def render(self):
        parts = []

        if self.eventID is not None:
            parts.append(u"id: {0}".format(self.eventID))

        if self.eventClass is not None:
            parts.append(u"event: {0}".format(self.eventClass))

        if self.retry is not None:
            parts.append(u"retry: {0:d}".format(self.retry))

        parts.extend(
            u"data: {}".format(line) for line in self.message.split(u"\n")
        )

        return (u"\r\n".join(parts) + u"\r\n\r\n")



@implementer(ILogObserver)
class DataStoreEventSourceLogObserver(object):
    """
    Observer for events related to any updates to the data store.
    """
    log = Logger()


    def __init__(self):
        self._listeners = set()
        self._events = deque(maxlen=1000)
        self._start = time()
        self._counter = 0


    def addListener(self, listener, lastEventID=None):
        self.log.debug("Adding listener: {listener}", listener=listener)

        self._playback(listener, lastEventID)

        self._listeners.add(listener)


    def removeListener(self, listener):
        self.log.debug("Removing listener: {listener}", listener=listener)

        self._listeners.add(listener)


    def _transmogrify(self, loggerEvent, eventID):
        """
        Convert a logger event into an EventSource event.
        """
        if "storeWriteClass" not in loggerEvent:
            # Not a data store event
            return None

        eventClass = loggerEvent.get("storeWriteClass", None)

        if eventClass is None:
            return

        elif eventClass is Incident:
            incident = loggerEvent.get("incident", None)

            if incident is None:
                incidentNumber = loggerEvent.get("incidentNumber", None)
            else:
                incidentNumber = incident.number

            if incidentNumber is None:
                self.log.critical(
                    "Unable to determine incident number from store event: "
                    "{event}",
                    event=loggerEvent,
                )
                return

            message = dict(incident_number=incidentNumber)

        else:
            self.log.debug(
                "Unknown data store event class: {eventClass}",
                eventClass=eventClass
            )
            return

        eventSourceEvent = Event(
            eventID=eventID,
            eventClass=eventClass.__name__,
            message=textFromJSON(message),
        )
        return eventSourceEvent


    def _playback(self, listener, lastEventID):
        if lastEventID is None:
            return

        observerID, counter = lastEventID.split(":")

        if observerID != str(id(self)):
            # lastEventID came from a different observer
            counter = 0

        for eventCounter, event in self.events:
            if eventCounter >= counter:
                listener.write(event.render().encode("utf-8"))


    def _publish(self, eventSourceEvent, eventID):
        eventText = eventSourceEvent.render().encode("utf-8")

        for listener in tuple(self._listeners):
            try:
                listener.write(eventText)
            except Exception:
                self.log.failure(
                    "Unable to publish to EventSource listener: {listener}",
                    listener=listener
                )
                self.removeListener(listener)

        self._events.append((self._counter, eventSourceEvent))


{'fraction_non_alphanumeric': 0.09175223583968202, 'fraction_numerical': 0.003754002429060395, 'mean_word_length': 2.139688041594454, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '6243665', 'n_tokens_mistral': 2428, 'n_tokens_neox': 2317, 'n_words': 703}