<commit_before>"""
ZeroMQ PUB-SUB wrappers.
"""
from zmq.core import constants

from twisted.internet import defer

from txzmq.connection import ZmqConnection


class ZmqXREQConnection(ZmqConnection):
    """
    A XREQ connection.
    """
    socketType = constants.XREQ

    def __init__(self, factory, *endpoints):
        ZmqConnection.__init__(self, factory, *endpoints)
        self._requests = {}

    def _getNextId(self):
        """
        Returns an unique id.
        """
        raise NotImplementedError(self)

    def sendMsg(self, *message_parts):
        """
        Send L{message} with specified L{tag}.

        @param message_parts: message data
        @type message: C{tuple}
        """
        d = defer.Deferred()
        message_id = self._getNextId()
        self._requests[message_id] = d
        self.send([message_id, ''] + list(message_parts))
        return d

    def messageReceived(self, message):
        """
        Called on incoming message from ZeroMQ.

        @param message: message data
        """
        msg_id, _, msg = message[0], message[1], message[2:]
        d = self._requests.pop(msg_id)
        d.callback(msg)


class ZmqXREPConnection(ZmqConnection):
    """
    A XREP connection.
    """
    socketType = constants.XREP

    def __init__(self, factory, *endpoints):
        ZmqConnection.__init__(self, factory, *endpoints)
        self._routing_info = {}  # keep track of routing info

    def reply(self, message_id, *message_parts):
        """
        Send L{message} with specified L{tag}.

        @param message_id: message uuid
        @type message_id: C{str}
        @param message: message data
        @type message: C{str}
        """
        routing_info = self._routing_info[message_id]
        self.send(routing_info + [message_id, ''] + list(message_parts))

    def messageReceived(self, message):
        """
        Called on incoming message from ZeroMQ.

        @param message: message data
        """
        i = message.index('')
        assert i > 0
        (routing_info, msg_id, payload) = (
            message[:i - 1], message[i - 1], message[i + 1:])
        msg_parts = payload[0:]<commit_msg>Cleanup routing info when sending reply in XREP.<commit_after>"""
ZeroMQ PUB-SUB wrappers.
"""
from zmq.core import constants

from twisted.internet import defer

from txzmq.connection import ZmqConnection


class ZmqXREQConnection(ZmqConnection):
    """
    A XREQ connection.
    """
    socketType = constants.XREQ

    def __init__(self, factory, *endpoints):
        ZmqConnection.__init__(self, factory, *endpoints)
        self._requests = {}

    def _getNextId(self):
        """
        Returns an unique id.
        """
        raise NotImplementedError(self)

    def sendMsg(self, *message_parts):
        """
        Send L{message} with specified L{tag}.

        @param message_parts: message data
        @type message: C{tuple}
        """
        d = defer.Deferred()
        message_id = self._getNextId()
        self._requests[message_id] = d
        self.send([message_id, ''] + list(message_parts))
        return d

    def messageReceived(self, message):
        """
        Called on incoming message from ZeroMQ.

        @param message: message data
        """
        msg_id, _, msg = message[0], message[1], message[2:]
        d = self._requests.pop(msg_id)
        d.callback(msg)


class ZmqXREPConnection(ZmqConnection):
    """
    A XREP connection.
    """
    socketType = constants.XREP

    def __init__(self, factory, *endpoints):
        ZmqConnection.__init__(self, factory, *endpoints)
        self._routing_info = {}  # keep track of routing info

    def reply(self, message_id, *message_parts):
        """
        Send L{message} with specified L{tag}.

        @param message_id: message uuid
        @type message_id: C{str}
        @param message: message data
        @type message: C{str}
        """
        routing_info = self._routing_info.pop(message_id)
        self.send(routing_info + [message_id, ''] + list(message_parts))

    def messageReceived(self, message):
        """
        Called on incoming message from ZeroMQ.

        @param message: message data
        """
        i = message.index('')
        assert i > 0
        (routing_info, msg_id, payload) = (
            message[:i - 1], message[i - 1], message[i + 1:])
        msg_parts = payload[0:]
{'fraction_non_alphanumeric': 0.11370328425821065, 'fraction_numerical': 0.00362400906002265, 'mean_word_length': 2.325301204819277, 'pattern_counts': {'":': 0, '<': 3, '<?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': '3580457', 'n_tokens_mistral': 1337, 'n_tokens_neox': 1296, 'n_words': 330}