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

"""ION messaging endpoints"""

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

from pyon.net.endpoint import ProcessRPCClient, ProcessRPCServer, Publisher, Subscriber
from pyon.public import CFG

class ProcessPublisher(Publisher):
    def __init__(self, process=None, **kwargs):
        self._process = process
        Publisher.__init__(self, **kwargs)

class EventPublisher(ProcessPublisher):
    """
    Class for process event publication - place holder
    """
    XS = CFG.exchanges.ioncore.name
    XP = CFG.exchanges.ioncore.exchange_points.events.name

class StreamPublisher(ProcessPublisher):
    """
    Data management abstraction of EndPoint layer for publishing messages to a stream
    TODO: Could be an object that uses a publisher?
    """

    XS = CFG.exchanges.ioncore.name
    XP = CFG.exchanges.ioncore.exchange_points.science_data.name

    def __init__(self, stream_route, **kwargs):
        self._stream_route = stream_route

        ProcessPublisher.__init__(self, name=stream_route.exchange_name, **kwargs)
        # @TODO Cant distinguish my exchange name from the to_name in the endpoint layer. Need a concept of FROM

class StreamPublisherRegistrar(object):
    """
    A Data Management level object for creating a publisher for a stream
    This object manages registration of publishers for different streams and creates the abstracted endpoint with the
    publish method
    """

    def __init__(self, process):
        """
        Use the process's exchange name to publish messages to a stream
        """
        self.process = process
        self.exchange_name = process.id
        self.pubsub_client = process.clients.pubsub_management

    def create_publisher(self, stream_id):
        """
        Call pubsub service to register this exchange name (endpoint) to publish on a particular stream
        Return a stream publisher object to publish (send) messages on a particular stream
        """

        # Call the pubsub service to register the exchange name as a publisher for this stream
        stream_route = self.pubsub_client.register_producer(self.exchange_name, stream_id)

        # Create the Stream publisher, ready to publish messages to the stream
        return StreamPublisher(stream_route, process, node=process.container.node)

class ProcessSubscriber(Subscriber):
    def __init__(self, process=None, **kwargs):
        self._process = process
        Subscriber.__init__(self, **kwargs)

class EventSubscriber(ProcessSubscriber):
    """
    Event Subscriber class for a process - place holder
    """


class StreamSubscriber(ProcessSubscriber):
    """
    Data management abstraction of the subscriber endpoint
    """
    def __init__(self, subscriber, subscription_id=None, **kwargs):
        ProcessSubscriber.__init__(self, **kwargs)
        self._subscriber = subscriber
        self._subscription_id = subscription_id

    def start(self):
        """
        Start consuming from the queue
        """
        # I now believe this is incorrect - how can this be implemented properly?
        channel = self._subscriber_endpoint.channel
        self._subscriber_endpoint.attach_channel(channel)
        # what should the correct behavior be for already attached?

    def stop(self):
        """
        Stop consuming from the queue
        """
        # @TODO - not implemented in the endpoint layer yet.

class StreamSubscriberRegistrar(object):
    """
    Class to create and register subscriptions in the pubsub service, create a StreamSubscriber
    """

    def __init__(self, process):
        self.process = process
        self.pubsub_client = process.clients.pubsub_management

    def subscribe(self, exchange_name='', callback=None, query=None):
        """
        This method creates a new subscriber, a new exchange_name if it does not already exist, and a new subscription
        if a query is provided.
        """

        # @TODO what about node - do I need to pass that?
        # @TODO what about an anonymous exchange name - how can we allow that?
        # Is there automatically a binding for the exchange name when it is created? We want the pubsub service to
        # control routes for this exchange name
        # Will this No-op if the exhange name already exists? That is what I intended...
        s = Subscriber(callback=callback, name=exchange_name)

        # @TODO - Is the subscriber already attached - is that the correct behavior?

        sub_id = None
        if query is not None:
            # Call the pubsub service to create a subscription if a query is provided
            subscription = IonObject("Subscription", {'exchange_name':exchange_name,'query':query})
            sub_id = self.pubsub_client.subscribe(subscription)

        return StreamSubscriber(s, sub_id, self.process)

    def activate_subscription(self, stream_subscriber):
        """
        Call the pubsub service to activate the subscription
        """
        self.pubsub_client.activate_subscription(stream_subscriber._subscription_id)

    def deactivate_subscription(self, stream_subscriber):
        """
        Call the pubsub service to deactivate the subscription
        """
        self.pubsub_client.deactivate_subscription(stream_subscriber._subscription_id)


<commit_msg>implementation is complete. Testing tomorrow.<commit_after>#!/usr/bin/env python

"""ION messaging endpoints"""
from pyon.util import log

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

from pyon.net.endpoint import ProcessRPCClient, ProcessRPCServer, Publisher, Subscriber
from pyon.public import CFG, IonObject
from pyon.util.log import log
from pyon.net.channel import PubChannel, SubscriberChannel, ChannelError
from pyon.util.async import  spawn
from interface.services.dm.ipubsub_management_service import PubsubManagementServiceProcessClient

class ProcessPublisher(Publisher):
    def __init__(self, process=None, **kwargs):
        self._process = process
        Publisher.__init__(self, **kwargs)


class PublisherError(StandardError):
    """
    An exception class for errors in the subscriber
    """


class StreamPublisher(ProcessPublisher):
    """
    Data management abstraction of EndPoint layer for publishing messages to a stream
    """

    class NoDeclarePublisherChannel(PubChannel):

        """
        # Once EMS exists - remove the declare!
        def _declare_exchange_point(self, xp):
            log.debug("StreamPublisher passing on _declare_exchange_point: %s", xp)
        """
    channel_type = NoDeclarePublisherChannel

    '''
    def __init__(self, **kwargs):
        """
        @param stream_route is a stream_route object
        @param process is the publishing process
        @param node is cc.node
        """

        self._stream_route = stream_route


        ProcessPublisher.__init__(self, **kwargs)
    '''


class StreamPublisherRegistrar(object):
    """
    A Data Management level object for creating a publisher for a stream
    This object manages registration of publishers for different streams and creates the abstracted endpoint with the
    publish method
    """

    def __init__(self, process=None, node=None):
        """
        Use the process's exchange name to publish messages to a stream
        """
        self.process = process
        self.exchange_name = process.id
        self.node = node
        self.pubsub_client = PubsubManagementServiceProcessClient(process=process, node=node)

        xs_dot_xp = CFG.core_xps.science_data
        try:
            self.XS, self.XP = xs_dot_xp.split('.')
        except ValueError:
            raise PublisherError('Invalid CFG for core_xps.science_data: "%s"; must have "xs.xp" structure' % xs_dot_xp)


    def create_publisher(self, stream_id):
        """
        Call pubsub service to register this exchange name (endpoint) to publish on a particular stream
        Return a stream publisher object to publish (send) messages on a particular stream
        """

        # Call the pubsub service to register the exchange name as a publisher for this stream
        stream_route = self.pubsub_client.register_producer(self.exchange_name, stream_id)

        # Create the Stream publisher, ready to publish messages to the stream
        return StreamPublisher(name=(self.XP, stream_route.exchange_name), process=self.process, node=self.node)



class SubscriberError(StandardError):
    """
    An exception class for errors in the subscriber
    """


class ProcessSubscriber(Subscriber):
    def __init__(self, process=None, **kwargs):
        self._process = process
        Subscriber.__init__(self, **kwargs)


class StreamSubscriber(ProcessSubscriber):
    """
    Data management abstraction of the subscriber endpoint
    """
    class NoBindSubscriberChannel(SubscriberChannel):

        def _bind(self, binding):
            log.debug("StreamSubscriber passing on _bind: %s", binding)

        """
        # Once EMS exists - remove the declare!
        def _declare_exchange_point(self, xp):
            log.debug("StreamSubscriber passing on _declare_exchange_point: %s", xp)
        """

    channel_type = NoBindSubscriberChannel


    def __init__(self, subscription_id=None, **kwargs):
        """
        @param stream_route is a stream_route object
        @param Process is the subscribing process
        @param node is cc.node
        """
        ProcessSubscriber.__init__(self, **kwargs)
        self.subscription_id = subscription_id

    def start(self):
        """
        Start consuming from the queue
        """
        if hasattr(self, '_chan'):
            try:
                self._chan.start_consume()
            except ChannelError:
                log.info('Subscriber is already started')

        else:
            self.gl = spawn(self.listen)


    def stop(self):
        """
        Stop consuming from the queue
        """

        if hasattr(self, '_chan'):
            self._chan.stop_consume()
        else:

            raise SubscriberError('Can not stop the subscriber before it is started')

    def close(self):
        self._chan.close()
        self.gl.join()

class StreamSubscriberRegistrarError(StandardError):
    """
    Error class for the StreamSubscriberRegistrar
    """

class StreamSubscriberRegistrar(object):
    """
    Class to create and register subscriptions in the pubsub service, create a StreamSubscriber
    """

    def __init__(self, process=None, node=None):
        self.process = process
        self.node = node
        self.pubsub_client = PubsubManagementServiceProcessClient(process=process, node=node)

        self._subscriber_cnt = 0

        xs_dot_xp = CFG.core_xps.science_data
        try:
            self.XS, self.XP = xs_dot_xp.split('.')
        except ValueError:
            raise PublisherError('Invalid CFG for core_xps.science_data: "%s"; must have "xs.xp" structure' % xs_dot_xp)


    def subscribe(self, exchange_name='', callback=None, query=None):
        """
        This method creates a new subscriber, a new exchange_name if it does not already exist, and a new subscription
        if a query is provided.
        """

        if not exchange_name:
            # if not create a new one based on the process id
            exchange_name =  '%s_subscriber_%d' % (self.process.id, self._subscriber_cnt)
            self._subscriber_cnt += 1

            if query is None:
                # one or the other must be provided
                raise StreamSubscriberRegistrarError('You can not register a new subscriber without a name or a query!')

        sub_id = None
        if query is not None:
            # Call the pubsub service to create a subscription if a query is provided
            subscription = IonObject("Subscription", {'exchange_name':exchange_name,'query':query})
            sub_id = self.pubsub_client.create_subscription(subscription)

        return StreamSubscriber(subscription_id=sub_id, name=(self.XP, exchange_name), process=self.process, callback=callback, node=self.node)

    def activate_subscription(self, stream_subscriber):
        """
        Call the pubsub service to activate the subscription
        """
        if stream_subscriber._subscription_id is None:
            raise StreamSubscriberRegistrarError('You can not activate a subscription if you do not own the subscription')
        self.pubsub_client.activate_subscription(stream_subscriber._subscription_id)

    def deactivate_subscription(self, stream_subscriber):
        """
        Call the pubsub service to deactivate the subscription
        """
        if stream_subscriber._subscription_id is None:
            raise StreamSubscriberRegistrarError('You can not deactivate a subscription if you do not own the subscription')
        self.pubsub_client.deactivate_subscription(stream_subscriber._subscription_id)



{'fraction_non_alphanumeric': 0.07666045934202359, 'fraction_numerical': 0.0004655493482309125, 'mean_word_length': 2.912871888281724, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, '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': '5284492', 'n_tokens_mistral': 3421, 'n_tokens_neox': 3258, 'n_words': 1091}