<commit_before>import logging
import pickle
import time
import json

try:
    import kafka
except ImportError:
    kafka = None

from .pubsub_manager import PubSubManager

logger = logging.getLogger('socketio')


class KafkaManager(PubSubManager):  # pragma: no cover
    """Kafka based client manager.

    This class implements a Kafka backend for event sharing across multiple
    processes.

    To use a Kafka backend, initialize the :class:`Server` instance as
    follows::

        url = 'kafka://hostname:port/0'
        server = socketio.Server(client_manager=socketio.KafkaManager(url))

    :param url: The connection URL for the Kafka server. For a default Kafka
                store running on the same host, use ``kafka://``.
    :param channel: The channel name (topic) on which the server sends and receives
                    notifications. Must be the same in all the servers.
    :param write_only: If set ot ``True``, only initialize to emit events. The
                       default of ``False`` initializes the class for emitting
                       and receiving.
    """
    name = 'kafka'

    def __init__(self, url='kafka://localhost:9092/0', channel='socketio',
                 write_only=False):
        if kafka is None:
            raise RuntimeError('kafka-python package is not installed '
                               '(Run "pip install kafka-python" in your '
                               'virtualenv).')
        self.kafka_url = url
        self.producer = kafka.KafkaProducer(bootstrap_servers='localhost:9092')

        super(KafkaManager, self).__init__(channel=channel,
                                           write_only=write_only)

    def initialize(self):
        super(KafkaManager, self).initialize()
        monkey_patched = True
        if self.server.async_mode == 'eventlet':
            from eventlet.patcher import is_monkey_patched
            monkey_patched = is_monkey_patched('socket')
        elif 'gevent' in self.server.async_mode:
            from gevent.monkey import is_module_patched
            monkey_patched = is_module_patched('socket')
        if not monkey_patched:
            raise RuntimeError(
                'Kafka requires a monkey patched socket library to work '
                'with ' + self.server.async_mode)


    def _publish(self, data):
        self.producer.send(self.channel, value=pickle.dumps(data))
        self.producer.flush()


    def _kafka_listen(self):
        self.consumer = kafka.KafkaConsumer(self.channel, bootstrap_servers='localhost:9092')
        for message in self.consumer:
            yield message


    def _listen(self):
        for message in self._kafka_listen():
            if message.topic == self.channel:
                yield pickle.loads(message.value)<commit_msg>Use the provided url in KafkaManager and remove the method 'initialize'<commit_after>import logging
import pickle
import time
import json

try:
    import kafka
except ImportError:
    kafka = None

from .pubsub_manager import PubSubManager

logger = logging.getLogger('socketio')


class KafkaManager(PubSubManager):  # pragma: no cover
    """Kafka based client manager.

    This class implements a Kafka backend for event sharing across multiple
    processes.

    To use a Kafka backend, initialize the :class:`Server` instance as
    follows::

        url = 'kafka://hostname:port'
        server = socketio.Server(client_manager=socketio.KafkaManager(url))

    :param url: The connection URL for the Kafka server. For a default Kafka
                store running on the same host, use ``kafka://``.
    :param channel: The channel name (topic) on which the server sends and receives
                    notifications. Must be the same in all the servers.
    :param write_only: If set ot ``True``, only initialize to emit events. The
                       default of ``False`` initializes the class for emitting
                       and receiving.
    """
    name = 'kafka'

    def __init__(self, url='kafka://localhost:9092', channel='socketio',
                 write_only=False):
        if kafka is None:
            raise RuntimeError('kafka-python package is not installed '
                               '(Run "pip install kafka-python" in your '
                               'virtualenv).')

        super(KafkaManager, self).__init__(channel=channel,
                                           write_only=write_only)

        self.kafka_url = url[8:] if url != 'kafka://' else 'localhost:9092'
        self.producer = kafka.KafkaProducer(bootstrap_servers=self.kafka_url)
        self.consumer = kafka.KafkaConsumer(self.channel,
                                            bootstrap_servers=self.kafka_url)


    def _publish(self, data):
        self.producer.send(self.channel, value=pickle.dumps(data))
        self.producer.flush()


    def _kafka_listen(self):
        for message in self.consumer:
            yield message


    def _listen(self):
        for message in self._kafka_listen():
            if message.topic == self.channel:
                yield pickle.loads(message.value)
{'fraction_non_alphanumeric': 0.0833008193523215, 'fraction_numerical': 0.004486929379633242, 'mean_word_length': 2.2449367088607595, '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': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5906517', 'n_tokens_mistral': 1392, 'n_tokens_neox': 1336, 'n_words': 417}