<commit_before>import json
import logging
import asyncio

import asyncio_redis as redis
from vyked.utils.log import log


class PubSubHandler:
    _logger = logging.getLogger(__name__)

    def __init__(self, redis_host, redis_port):
        self._redis_host = redis_host
        self._redis_port = redis_port
        self._conn = None

    @asyncio.coroutine
    def connect(self):
        self._conn = yield from self._get_conn()
        return self._conn

    @asyncio.coroutine
    @log(logger=_logger.error)
    def publish(self, service, version, endpoint, payload:str):
        if self._conn is not None:
            try:
                yield from self._conn.publish(self._get_key(service, version, endpoint), json.dumps(payload))
                return True
            except redis.Error as e:
                self._logger.error('Publish failed with error %s', repr(e))
        return False

    @asyncio.coroutine
    def subscribe(self, endpoints, handler):
        connection = yield from self._get_conn()
        subscriber = yield from connection.start_subscribe()
        channels = [self._get_key(endpoint['service'], endpoint['version'], endpoint['endpoint']) for endpoint in
                    endpoints]
        yield from subscriber.subscribe(channels)
        while True:
            payload = yield from subscriber.next_published()
            service, version, endpoint = payload.channel.split('.')
            handler(service, int(version), endpoint, payload.value)
        return False

    @staticmethod
    def _get_key(service, version, endpoint):
        return '{}.{}.{}'.format(service, version, endpoint)

    def _get_conn(self):
        return (yield from redis.Connection.create(self._redis_host, self._redis_port, auto_reconnect=True))<commit_msg>Refactoring pubsub class<commit_after>import logging
import asyncio

import asyncio_redis as redis

from vyked.utils.log import log


class PubSubHandler:
    """
    Pub sub handler which uses redis.
    Can be used to publish an event or subscribe to a list of endpoints
    """
    _logger = logging.getLogger(__name__)

    def __init__(self, redis_host, redis_port):
        """
        Create in instance of Pub Sub handler
        :param str redis_host: Redis Host address
        :param redis_port: Redis port number
        """
        self._redis_host = redis_host
        self._redis_port = redis_port
        self._conn = None

    @asyncio.coroutine
    def connect(self):
        """
        Connect to the redis server and return the connection
        :return:
        """
        self._conn = yield from self._get_conn()
        return self._conn

    @asyncio.coroutine
    @log(logger=_logger.error)
    def publish(self, endpoint: str, payload: str):
        """
        Publish to an endpoint.
        :param str endpoint: Key by which the endpoint is recognised.
                         Subscribers will use this key to listen to events
        :param str payload: Payload to publish with the event
        :return: A boolean indicating if the publish was successful
        """
        if self._conn is not None:
            try:
                yield from self._conn.publish(endpoint, payload)
                return True
            except redis.Error as e:
                self._logger.error('Publish failed with error %s', repr(e))
        return False

    @asyncio.coroutine
    def subscribe(self, endpoints: list, handler):
        """
        Subscribe to a list of endpoints
        :param endpoints: List of endpoints the subscribers is interested to subscribe to
        :type endpoints: list
        :param handler: The callback to call when a particular event is published.
                        Must take two arguments, a channel to which the event was published
                        and the payload.
        :return:
        """
        connection = yield from self._get_conn()
        subscriber = yield from connection.start_subscribe()
        yield from subscriber.subscribe(endpoints)
        while True:
            payload = yield from subscriber.next_published()
            handler(payload.channel, payload.value)
        return False

    def _get_conn(self):
        return (yield from redis.Connection.create(self._redis_host, self._redis_port, auto_reconnect=True))
{'fraction_non_alphanumeric': 0.07799442896935933, 'fraction_numerical': 0.0, 'mean_word_length': 2.348096348096348, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '2192121', 'n_tokens_mistral': 1148, 'n_tokens_neox': 1112, 'n_words': 369}