<commit_before>#!/usr/bin/env python

"""Events and Notifications"""

__author__ = '<NAME> <<EMAIL>>, <NAME>'
__license__ = 'Apache 2.0'

from gevent import event as gevent_event

from pyon.core import bootstrap
from pyon.core.exception import BadRequest, IonException
from pyon.datastore.datastore import DataStore
from pyon.net.endpoint import Publisher, Subscriber
from pyon.util.async import spawn
from pyon.util.containers import get_ion_ts
from pyon.util.log import log

from interface.objects import Event

# @TODO: configurable
EVENTS_XP = "pyon.events"
EVENTS_XP_TYPE = "topic"

PERSIST_ON_PUBLISH = False


def get_events_exchange_point():
    return "%s.%s" % (bootstrap.get_sys_name(), EVENTS_XP)


class EventError(IonException):
    status_code = 500


class EventPublisher(Publisher):

    def __init__(self, event_type=None, xp=None, **kwargs):
        """
        Constructs a publisher of events for a specific type.

        @param  event_type  The name of the event type object
        @param  xp          Exchange (AMQP) name, can be none, will use events default.
        """

        self.event_type = event_type

        if bootstrap.container_instance and getattr(bootstrap.container_instance, 'event_repository', None):
            self.event_repo = bootstrap.container_instance.event_repository
        else:
            self.event_repo = None

        # generate an exchange name to publish events to
        xp = xp or get_events_exchange_point()
        name = (xp, None)

        Publisher.__init__(self, to_name=name, **kwargs)

    def _topic(self, event_type, origin, base_types=None, sub_type=None, origin_type=None):
        """
        Builds the topic that this event should be published to.
        """
        assert event_type and origin
        base_types = base_types or []
        base_str = ".".join(reversed(base_types))
        sub_type = sub_type or "_"
        origin_type = origin_type or "_"
        routing_key = "%s.%s.%s.%s.%s" % (base_str, event_type, sub_type, origin_type, origin)
        return routing_key

    def _create_event(self, event_type=None, **kwargs):
        event_type = event_type or self.event_type
        assert event_type

        if 'ts_created' not in kwargs:
            kwargs['ts_created'] = get_ion_ts()

        event_msg = bootstrap.IonObject(event_type, **kwargs)
        event_msg.base_types = event_msg._get_extends()

        # Would like to validate here but blows up if an object is provided where a dict is declared
        #event_msg._validate()

        return event_msg

    def _publish_event(self, event_msg, origin, event_type=None):
        event_type = event_type or self.event_type or event_msg._get_type()
        assert origin and event_type

        topic = self._topic(event_type, origin, base_types=event_msg.base_types,
            sub_type=event_msg.sub_type, origin_type=event_msg.origin_type)
        to_name = (self._send_name.exchange, topic)
        log.trace("Publishing event message to %s", to_name)

        try:
            self.publish(event_msg, to_name=to_name)
        except Exception as ex:
            log.exception("Failed to publish event '%s'" % (event_msg))
            return False

        try:
            # store published event but only if we specified an event_repo
            if PERSIST_ON_PUBLISH and self.event_repo:
                self.event_repo.put_event(event_msg)
        except Exception as ex:
            log.exception("Failed to store published event '%s'" % (event_msg))
            return False

        return True

    def publish_event(self, origin=None, event_type=None, **kwargs):
        """
        Publishes an event of given type for the given origin. Event_type defaults to an
        event_type set when initializing the EventPublisher. Other kwargs fill out the fields
        of the event. This operation will log any errors occuring during event creation,
        but will not fail with an exception.
        @param origin     the origin field value
        @param event_type the event type (defaults to the EventPublisher's event_typeif set)
        @param kwargs     additional event fields
        @retval  bool  indicating success of this operation
        """
        event_msg = self._create_event(origin=origin, event_type=event_type, **kwargs)
        success = self._publish_event(event_msg, origin=origin)
        return success


class BaseEventSubscriberMixin(object):
    """
    A mixin class for Event subscribers to facilitate inheritance.

    EventSubscribers must come in both standard and process level versions, which
    rely on common base code. It is difficult to multiple inherit due to both of
    them sharing a base class, so this mixin is preferred.
    """

    @staticmethod
    def _topic(event_type, origin, sub_type=None, origin_type=None):
        """
        Builds the topic that this event should be published to.
        If either side of the event_id.origin pair are missing, will subscribe to anything.
        """
        if event_type == "Event":
            event_type = "Event.#"
        elif event_type:
            event_type = "#.%s.#" % event_type
        else:
            event_type = "#"

        sub_type = sub_type or "*.#"
        origin_type = origin_type or "*"<commit_msg>OOIION-614: add ionobject parameter to EventPublisher:publish_event method<commit_after>#!/usr/bin/env python

"""Events and Notifications"""

__author__ = '<NAME> <<EMAIL>>, <NAME>'
__license__ = 'Apache 2.0'

from gevent import event as gevent_event

from pyon.core import bootstrap
from pyon.core.exception import BadRequest, IonException
from pyon.datastore.datastore import DataStore
from pyon.net.endpoint import Publisher, Subscriber
from pyon.util.async import spawn
from pyon.util.containers import get_ion_ts
from pyon.util.log import log

from interface.objects import Event

# @TODO: configurable
EVENTS_XP = "pyon.events"
EVENTS_XP_TYPE = "topic"

PERSIST_ON_PUBLISH = False


def get_events_exchange_point():
    return "%s.%s" % (bootstrap.get_sys_name(), EVENTS_XP)


class EventError(IonException):
    status_code = 500


class EventPublisher(Publisher):

    def __init__(self, event_type=None, xp=None, **kwargs):
        """
        Constructs a publisher of events for a specific type.

        @param  event_type  The name of the event type object
        @param  xp          Exchange (AMQP) name, can be none, will use events default.
        """

        self.event_type = event_type

        if bootstrap.container_instance and getattr(bootstrap.container_instance, 'event_repository', None):
            self.event_repo = bootstrap.container_instance.event_repository
        else:
            self.event_repo = None

        # generate an exchange name to publish events to
        xp = xp or get_events_exchange_point()
        name = (xp, None)

        Publisher.__init__(self, to_name=name, **kwargs)

    def _topic(self, event_type, origin, base_types=None, sub_type=None, origin_type=None):
        """
        Builds the topic that this event should be published to.
        """
        assert event_type and origin
        base_types = base_types or []
        base_str = ".".join(reversed(base_types))
        sub_type = sub_type or "_"
        origin_type = origin_type or "_"
        routing_key = "%s.%s.%s.%s.%s" % (base_str, event_type, sub_type, origin_type, origin)
        return routing_key

    def _create_event(self, event_type=None, ionobject=None, **kwargs):
        if ionobject is None:
            event_type = event_type or self.event_type
            assert event_type

            if 'ts_created' not in kwargs:
                kwargs['ts_created'] = get_ion_ts()

            event_msg = bootstrap.IonObject(event_type, **kwargs)
        event_msg.base_types = event_msg._get_extends()

        # Would like to validate here but blows up if an object is provided where a dict is declared
        #event_msg._validate()

        return event_msg

    def _publish_event(self, event_msg, origin, event_type=None):
        event_type = event_type or self.event_type or event_msg._get_type()
        assert origin and event_type

        topic = self._topic(event_type, origin, base_types=event_msg.base_types,
            sub_type=event_msg.sub_type, origin_type=event_msg.origin_type)
        to_name = (self._send_name.exchange, topic)
        log.trace("Publishing event message to %s", to_name)

        try:
            self.publish(event_msg, to_name=to_name)
        except Exception as ex:
            log.exception("Failed to publish event '%s'" % (event_msg))
            return False

        try:
            # store published event but only if we specified an event_repo
            if PERSIST_ON_PUBLISH and self.event_repo:
                self.event_repo.put_event(event_msg)
        except Exception as ex:
            log.exception("Failed to store published event '%s'" % (event_msg))
            return False

        return True

    def publish_event(self, origin=None, event_type=None, ionobject=None, **kwargs):
        """
        Publishes an event of given type for the given origin. Event_type defaults to an
        event_type set when initializing the EventPublisher. Other kwargs fill out the fields
        of the event. This operation will log any errors occuring during event creation,
        but will not fail with an exception.
        @param origin     the origin field value
        @param event_type the event type (defaults to the EventPublisher's event_typeif set)
        @param kwargs     additional event fields
        @retval  bool  indicating success of this operation
        """
        event_msg = self._create_event(origin=origin, event_type=event_type, ionobject=ionobject, **kwargs)
        success = self._publish_event(event_msg, origin=origin)
        return success


class BaseEventSubscriberMixin(object):
    """
    A mixin class for Event subscribers to facilitate inheritance.

    EventSubscribers must come in both standard and process level versions, which
    rely on common base code. It is difficult to multiple inherit due to both of
    them sharing a base class, so this mixin is preferred.
    """

    @staticmethod
    def _topic(event_type, origin, sub_type=None, origin_type=None):
        """
        Builds the topic that this event should be published to.
        If either side of the event_id.origin pair are missing, will subscribe to anything.
        """
        if event_type == "Event":
            event_type = "Event.#"
        elif event_type:
            event_type = "#.%s.#" % event_type
        else:
            event_type = "#"

        sub_type = sub_type or "*.#"
        origin_type = origin_type or "*"
{'fraction_non_alphanumeric': 0.0786684024549005, 'fraction_numerical': 0.001208852519992561, 'mean_word_length': 2.7618048268625395, 'pattern_counts': {'":': 2, '<': 11, '<?xml version=': 0, '>': 11, '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': '7206332', 'n_tokens_mistral': 3137, 'n_tokens_neox': 2989, 'n_words': 975}