�
���gT � � � d dl Z d dlZ d dlZn
# e$ r dZY nw xY wddlmZ e j d� � Z G d� de� � ZdS )� N� )�
PubSubManager�socketioc �> � � e Zd ZdZdZ d
� fd� Zd� Zd� Zd � Z� xZ 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.
�kafka�kafka://localhost:9092r Fc �X �� t �t d� � �t � � � ||�� � t |t
� � r|gn|}d� |D � � | _ t j | j �� � | _ t j | j
| j �� � | _ d S )NzZkafka-python package is not installed (Run "pip install kafka-python" in your virtualenv).)�channel�
write_onlyc �2 � g | ]}|d k r
|dd� nd��S )zkafka://� Nzlocalhost:9092� )�.0�urls �f/home/asafur/pinokio/api/open-webui.git/app/env/lib/python3.11/site-packages/socketio/kafka_manager.py�
<listcomp>z)KafkaManager.__init__.<locals>.<listcomp>1 s? � � ,� ,� ,� "� '*�Z�&7�&7�3�q�r�r�7�7�=M� ,� ,� ,� )�bootstrap_servers) r � RuntimeError�super�__init__�
isinstance�str�
kafka_urls�
KafkaProducer�producer�
KafkaConsumerr �consumer)�selfr r r �urls� __class__s �r r zKafkaManager.__init__' s� �� � �=� � .� /� /�
/� ������Z��@�@�@�"�3��,�,�5��u�u�#��,� ,�&*�,� ,� ,��� �+�d�o�N�N�N��
� �+�D�L�>B�o�O� O� O��
�
�
r c � � | j � | j t j |� � �� � | j � � � d S )N)�value)r �sendr �pickle�dumps�flush)r �datas r �_publishzKafkaManager._publish7 sC � � �
���4�<�v�|�D�/A�/A��B�B�B� �
��� � � � � r c # �$ K � | j E d {V �� d S �N)r )r s r �
_kafka_listenzKafkaManager._kafka_listen; s& � � � ��=� � � � � � � � � r c # � K � | � � � D ]-}|j | j k rt j |j � � V � �.d S r, )r- �topicr r& �loadsr$ )r �messages r �_listenzKafkaManager._listen> sS � � � ��)�)�+�+� 2� 2�G��}�� �,�,� �l�7�=�1�1�1�1�1�� 2� 2r )r r F)
�__name__�
__module__� __qualname__�__doc__�namer r* r- r2 �
__classcell__)r" s @r r r s� �� � � � � �� �, �D�=G� !�O� O� O� O� O� O� � � �!� !� !�2� 2� 2� 2� 2� 2� 2r r ) �loggingr& r � ImportError�pubsub_managerr � getLogger�loggerr r r r �<module>r> s� �� ����
�
�
�
���L�L�L�L��� � � � �E�E�E����� *� )� )� )� )� )� �� �:� &� &��32� 32� 32� 32� 32�=� 32� 32� 32� 32� 32s � ��