<commit_before>"""
The Zeroless module API.

.. data:: log

A global Logger object. To use it, just add an Handler object
and set an appropriate logging level.
"""

import zmq
import logging

from time import sleep
from copy import deepcopy
from warnings import warn
from functools import partial

log = logging.getLogger(__name__)
log.addHandler(logging.NullHandler())

def _check_valid_port_range(port):
    if port < 1024 or port > 65535:
        error = 'Port {0} is invalid, choose one between 1024 and 65535'
        error = error.format(port)
        log.error(error)
        raise ValueError(error)

def _check_valid_num_connections(socket_type, num_connections):
    if socket_type == zmq.PAIR and num_connections > 1:
        error = 'A client cannot connect more than once in the PAIR pattern'
        log.error(error)
        raise RuntimeError(error)

def _connect_zmq_sock(sock, ip, port):
    log.info('Connecting to {0} on port {1}'.format(ip, port))
    sock.connect('tcp://' + ip + ':' + str(port))

def _disconnect_zmq_sock(sock, ip, port):
    log.info('Disconnecting from {0} on port {1}'.format(ip, port))

    try:
        sock.disconnect('tcp://' + ip + ':' + str(port))
    except zmq.ZMQError:
        error = 'There was no connection to {0} on port {1}'.format(ip, port)
        log.exception(error)
        raise ValueError(error)

def _bind_zmq_sock(sock, port):
    log.info('Binding on port {0}'.format(port))

    try:
        sock.bind('tcp://*:' + str(port))
    except zmq.ZMQError:
        error = 'Port {0} is already in use'.format(port)
        log.exception(error)
        raise ValueError(error)

def _recv(sock):
    while True:
        frames = sock.recv_multipart()
        log.debug('Receiving: {0}'.format(frames))
        yield frames if len(frames) > 1 else frames[0]

class Sock:
    def __init__(self):
        pass

    def __sock(self, pattern):
        sock = zmq.Context().instance().socket(pattern)
        self._setup(sock)

        log.info('Ready...')

        return sock

    def __send_function(self, sock, topic=None):
        def _send(*data):
            log.debug('Sending: {0}'.format(data))

            try:
                sock.send_multipart(data)
            except TypeError:
                error = 'Data must be bytes, so try again'
                log.exception(error)
                raise TypeError(error)

        if sock.socket_type == zmq.PUB:
            if topic:
                return partial(_send, topic)
            else:
                return partial(_send, b'')

        return _send

    def __recv_generator(self, sock):
        return _recv(sock)

    # PubSub pattern
    def pub(self, topic=b''):
        """
        Returns a callable that can be used to transmit a message, with a given
        ``topic``, in a publisher-subscriber fashion. Note that the sender
        function has a ``print`` like signature, with an infinite number of
        arguments. Each one being a part of the complete message.

        :param topic: the topic that will be published to (default=b'')
        :type topic: bytes
        :rtype: function
        """
        if not isinstance(topic, bytes):
            error = 'Topic must be bytes'
            log.error(error)
            raise TypeError(error)

        sock = self.__sock(zmq.PUB)
        return self.__send_function(sock, topic)

    def sub(self, topics=(b'',)):
        """
        Returns an iterable that can be used to iterate over incoming messages,
        that were published with one of the topics specified in ``topics``. Note
        that the iterable returns as many parts as sent by subscribed publishers.

        :param topics: a list of topics to subscribe to (default=b'')
        :type topics: list of bytes
        :rtype: generator
        """
        sock = self.__sock(zmq.SUB)

        for topic in topics:
            if not isinstance(topic, bytes):
                error = 'Topics must be a list of bytes'
                log.error(error)
                raise TypeError(error)
            sock.setsockopt(zmq.SUBSCRIBE, topic)<commit_msg>Added an embed_topic parameter to turn topic sending, when using publishers, optional.<commit_after>"""
The Zeroless module API.

.. data:: log

A global Logger object. To use it, just add an Handler object
and set an appropriate logging level.
"""

import zmq
import logging

from time import sleep
from copy import deepcopy
from warnings import warn
from functools import partial

log = logging.getLogger(__name__)
log.addHandler(logging.NullHandler())

def _check_valid_port_range(port):
    if port < 1024 or port > 65535:
        error = 'Port {0} is invalid, choose one between 1024 and 65535'
        error = error.format(port)
        log.error(error)
        raise ValueError(error)

def _check_valid_num_connections(socket_type, num_connections):
    if socket_type == zmq.PAIR and num_connections > 1:
        error = 'A client cannot connect more than once in the PAIR pattern'
        log.error(error)
        raise RuntimeError(error)

def _connect_zmq_sock(sock, ip, port):
    log.info('Connecting to {0} on port {1}'.format(ip, port))
    sock.connect('tcp://' + ip + ':' + str(port))

def _disconnect_zmq_sock(sock, ip, port):
    log.info('Disconnecting from {0} on port {1}'.format(ip, port))

    try:
        sock.disconnect('tcp://' + ip + ':' + str(port))
    except zmq.ZMQError:
        error = 'There was no connection to {0} on port {1}'.format(ip, port)
        log.exception(error)
        raise ValueError(error)

def _bind_zmq_sock(sock, port):
    log.info('Binding on port {0}'.format(port))

    try:
        sock.bind('tcp://*:' + str(port))
    except zmq.ZMQError:
        error = 'Port {0} is already in use'.format(port)
        log.exception(error)
        raise ValueError(error)

def _recv(sock):
    while True:
        frames = sock.recv_multipart()
        log.debug('Receiving: {0}'.format(frames))
        yield frames if len(frames) > 1 else frames[0]

class Sock:
    def __init__(self):
        pass

    def __sock(self, pattern):
        sock = zmq.Context().instance().socket(pattern)
        self._setup(sock)

        log.info('Ready...')

        return sock

    def __send_function(self, sock, topic=None, embed_topic=False):
        def _send(*data):
            log.debug('Sending: {0}'.format(data))

            try:
                sock.send_multipart(data)
            except TypeError:
                error = 'Data must be bytes, so try again'
                log.exception(error)
                raise TypeError(error)

        if sock.socket_type == zmq.PUB and embed_topic:
            if topic:
                return partial(_send, topic)
            else:
                return partial(_send, b'')

        return _send

    def __recv_generator(self, sock):
        return _recv(sock)

    # PubSub pattern
    def pub(self, topic=b'', embed_topic=False):
        """
        Returns a callable that can be used to transmit a message, with a given
        ``topic``, in a publisher-subscriber fashion. Note that the sender
        function has a ``print`` like signature, with an infinite number of
        arguments. Each one being a part of the complete message.

        :param topic: the topic that will be published to (default=b'')
        :type topic: bytes
        :param embed_topic: set to embed the topic into published messages
                            (default=False)
        :type embed_topic bool
        :rtype: function
        """
        if not isinstance(topic, bytes):
            error = 'Topic must be bytes'
            log.error(error)
            raise TypeError(error)

        sock = self.__sock(zmq.PUB)
        return self.__send_function(sock, topic, embed_topic)

    def sub(self, topics=(b'',)):
        """
        Returns an iterable that can be used to iterate over incoming messages,
        that were published with one of the topics specified in ``topics``. Note
        that the iterable returns as many parts as sent by subscribed publishers.

        :param topics: a list of topics to subscribe to (default=b'')
        :type topics: list of bytes
        :rtype: generator
        """
        sock = self.__sock(zmq.SUB)

        for topic in topics:
            if not isinstance(topic, bytes):
                error = 'Topics must be a list of bytes'
                log.error(error)
                raise TypeError(error)
            sock.setsockopt(zmq.SUBSCRIBE, topic)
{'fraction_non_alphanumeric': 0.10089369708372531, 'fraction_numerical': 0.007525870178739417, 'mean_word_length': 2.304195804195804, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 9, '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': '1509318', 'n_tokens_mistral': 2513, 'n_tokens_neox': 2410, 'n_words': 819}