i* |ddlZ ddlZn #e$rdZYnwxYwddlmZejdZGddeZdS)N) PubSubManagersocketioc>eZdZdZdZ d fd ZdZd Zd ZxZ S) KafkaManagera}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://``. For a highly available deployment of Kafka, pass a list with all the connection URLs available in your cluster. :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 to ``True``, only initialize to emit events. The default of ``False`` initializes the class for emitting and receiving. A write-only instance can be used independently of the server to emit to clients from an external process. :param logger: a custom logger to log it. If not given, the server logger is used. :param json: An alternative JSON module to use for encoding and decoding packets. Custom json modules must have ``dumps`` and ``loads`` functions that are compatible with the standard library versions. This setting is only used when ``write_only`` is set to ``True``. Otherwise the JSON module configured in the server is used. kafkakafka://localhost:9092rFNc\ttdt||||t |t r|gn|}d|D|_tj|j|_tj |j |j|_ dS)NzZkafka-python package is not installed (Run "pip install kafka-python" in your virtualenv).)channel write_onlyloggerjsonc2g|]}|dkr |ddndS)zkafka://Nzlocalhost:9092).0urls A/usr/local/lib/python3.11/dist-packages/socketio/kafka_manager.py z)KafkaManager.__init__..;s?,,,"'*Z&7&73qrr77=M,,,)bootstrap_servers) r RuntimeErrorsuper__init__ isinstancestr kafka_urls KafkaProducerproducer KafkaConsumerr consumer)selfrr r r rurls __class__s rrzKafkaManager.__init__0s = .// / Z"  $ $ $#3,,5uu#,,&*,,,+doNNN +DL>BoOOO rc|j|j|j||jdS)N)value)rsendr rdumpsflush)r"datas r_publishzKafkaManager._publishAsG 4r=sLLLL EEE*)))))  : & &>$>$>$>$>$=>$>$>$>$>$s