Browse Source

formatted code

pull/840/head
sparkingdark 3 years ago
parent
commit
b2580ccd25
  1. 36
      src/socketio/kafka_manager.py

36
src/socketio/kafka_manager.py

@ -1,6 +1,6 @@
import logging import logging
import pickle import pickle
from typing import List
try: try:
import kafka import kafka
@ -9,7 +9,7 @@ except ImportError:
from .pubsub_manager import PubSubManager from .pubsub_manager import PubSubManager
logger = logging.getLogger('socketio') logger = logging.getLogger("socketio")
class KafkaManager(PubSubManager): # pragma: no cover class KafkaManager(PubSubManager): # pragma: no cover
@ -26,6 +26,8 @@ class KafkaManager(PubSubManager): # pragma: no cover
:param url: The connection URL for the Kafka server. For a default Kafka :param url: The connection URL for the Kafka server. For a default Kafka
store running on the same host, use ``kafka://``. store running on the same host, use ``kafka://``.
From now on you can pass a list of URLs to use
different Kafka servers.
:param channel: The channel name (topic) on which the server sends and :param channel: The channel name (topic) on which the server sends and
receives notifications. Must be the same in all the receives notifications. Must be the same in all the
servers. servers.
@ -33,23 +35,29 @@ class KafkaManager(PubSubManager): # pragma: no cover
default of ``False`` initializes the class for emitting default of ``False`` initializes the class for emitting
and receiving. and receiving.
""" """
name = 'kafka'
def __init__(self, url='kafka://localhost:9092',channel='socketio', name = "kafka"
write_only=False):
def __init__(
self, url="kafka://localhost:9092", channel="socketio", write_only=False
):
if kafka is None: if kafka is None:
raise RuntimeError('kafka-python package is not installed ' raise RuntimeError(
'(Run "pip install kafka-python" in your ' "kafka-python package is not installed "
'virtualenv).') '(Run "pip install kafka-python" in your '
"virtualenv)."
)
super(KafkaManager, self).__init__(channel=channel, write_only=write_only)
super(KafkaManager, self).__init__(channel=channel,
write_only=write_only)
urls = [url] if isinstance(url, str) else url urls = [url] if isinstance(url, str) else url
self.kafka_urls = [url[8:] if url != 'kafka://' else 'localhost:9092' for url in urls] self.kafka_urls = [
url[8:] if url != "kafka://" else "localhost:9092" for url in urls
]
self.producer = kafka.KafkaProducer(bootstrap_servers=self.kafka_urls) self.producer = kafka.KafkaProducer(bootstrap_servers=self.kafka_urls)
self.consumer = kafka.KafkaConsumer(self.channel, self.consumer = kafka.KafkaConsumer(
bootstrap_servers=self.kafka_urls) self.channel, bootstrap_servers=self.kafka_urls
)
def _publish(self, data): def _publish(self, data):
self.producer.send(self.channel, value=pickle.dumps(data)) self.producer.send(self.channel, value=pickle.dumps(data))

Loading…
Cancel
Save