<commit_before># -*- coding: utf-8 -*-
import zmq

from . import logger

__all__ = ['Client']


class Client(object):
    """
    :class:`carotte.Client` can send task and manage it.

    :param list workers: List of worker addresses
    :param int timeout: Socket timeout
    :param boolean reconnect: Auto reconnect socket

    >>> from carotte import Client
    >>> client = Client()
    >>> task = client.run_task('hello')
    >>> task.terminated
    >>> False
    >>> task.wait()
    >>> task.terminated
    >>> True
    >>> task.result
    >>> 'hello world'
    """
    def __init__(self, workers=["tcp://localhost:5550"], timeout=10, reconnect=True):
        self.workers = workers
        self.context = zmq.Context()
        self.socket = self.context.socket(zmq.REQ)
        self.socket.setsockopt(zmq.LINGER, 0)

        self.poller = zmq.Poller()
        self.poller.register(self.socket, zmq.POLLIN)
        self.timeout = timeout
        self.reconnect = reconnect

        for address in self.workers:
            logger.info('Connecting to %s ...' % address)
            self.socket.connect(address)

    def __connect_socket(self):
        for address in self.workers:
            logger.info('Reconnecting to %s ...' % address)
            self.socket.connect(address)

    def __send_pyobj(self, data):
        try:
            self.socket.send_pyobj(data)
        except zmq.error.ZMQError:
            if self.reconnect:
                self.__connect_socket()
            else:
                raise

    def __recv_pyobj(self):
        if self.poller.poll(self.timeout * 1000):
            r = self.socket.recv_pyobj()
            if not r.get('success', False):
                exception = r.get('exception', Exception('Unhandler exception'))
                raise exception
            return r.get('task')
        else:
            raise IOError('Socket timeout (%s)' % self.workers)

    def run_task(self, task_name, task_args=[], task_kwargs={}):
        """
        Run asynchronous task on a :class:`carotte.Worker`.

        :param string task_name: Name of task to execute
        :param list task_args: (optional) List of arguments to give to task
        :param dict task_kwargs: (optional) Dict of keyword arguments
                                 to give to task<commit_msg>Remove multi workers<commit_after># -*- coding: utf-8 -*-
import zmq

from . import logger

__all__ = ['Client']


class Client(object):
    """
    :class:`carotte.Client` can send task and manage it.

    :param list worker: Worker address
    :param int timeout: Socket timeout
    :param boolean reconnect: Auto reconnect socket

    >>> from carotte import Client
    >>> client = Client()
    >>> task = client.run_task('hello')
    >>> task.terminated
    >>> False
    >>> task.wait()
    >>> task.terminated
    >>> True
    >>> task.result
    >>> 'hello world'
    """
    def __init__(self, worker="tcp://localhost:5550", timeout=10, reconnect=True):
        self.worker = worker
        self.context = zmq.Context()
        self.socket = self.context.socket(zmq.REQ)
        self.socket.setsockopt(zmq.LINGER, 0)

        self.poller = zmq.Poller()
        self.poller.register(self.socket, zmq.POLLIN)
        self.timeout = timeout
        self.reconnect = reconnect

        logger.info('Connecting to %s ...' % self.worker)
        self.socket.connect(self.worker)

    def __connect_socket(self):
        logger.info('Reconnecting to %s ...' % self.worker)
        self.socket.connect(self.worker)

    def __send_pyobj(self, data):
        try:
            self.socket.send_pyobj(data)
        except zmq.error.ZMQError:
            if self.reconnect:
                self.__connect_socket()
            else:
                raise

    def __recv_pyobj(self):
        if self.poller.poll(self.timeout * 1000):
            r = self.socket.recv_pyobj()
            if not r.get('success', False):
                exception = r.get('exception', Exception('Unhandler exception'))
                raise exception
            return r.get('task')
        else:
            raise IOError('Socket timeout (%s)' % self.workers)

    def run_task(self, task_name, task_args=[], task_kwargs={}):
        """
        Run asynchronous task on a :class:`carotte.Worker`.

        :param string task_name: Name of task to execute
        :param list task_args: (optional) List of arguments to give to task
        :param dict task_kwargs: (optional) Dict of keyword arguments
                                 to give to task
{'fraction_non_alphanumeric': 0.1189165382074433, 'fraction_numerical': 0.005285179475886369, 'mean_word_length': 2.1173644474948525, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 63, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5677017', 'n_tokens_mistral': 1329, 'n_tokens_neox': 1278, 'n_words': 337}