0x1998 - MANAGER
Düzenlenen Dosya: accumulate.cpython-311.pyc
� ���i� � � � d dl Z d dlZd dlZd dlZd dlmZ d dlmZ d dlm Z m Z m Z d dl m Z mZmZ d dlmZ ee� � Z G d� de e� � ZdS ) � N)� getLogger)� inactivity)� Accumulatable� MessageType� Splittable)� MessageSink� MessageSource�expect)�recurring_checkc �& � � e Zd Zej j ZdZ ee j � dd� � � � Z ee j � dd� � � � Z e e f� fd� Zd� Zd� Zd � Zd � Z eej � � d efd �� � Zd � Z� xZS )� Accumulate�� �%IMUNIFY360_AGGREGATE_MESSAGES_TIMEOUT�< �*IMUNIFY360_AGGREGATE_SHUTDOWN_SEND_TIMEOUT�2 c � �� t � � j di |�� || _ || _ t j t � � | _ d S )N� )�super�__init__�_period�_shutdown_timeout� collections� defaultdict�list�_data)�self�period�shutdown_timeout�kwargs� __class__s ��W/opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/plugins/accumulate.pyr zAccumulate.__init__# sG �� � �����"�"�6�"�"�"� �� �!1�� � �,�T�2�2�� � � � c � �� K � || _ || _ | j dk rd n<|� t | j � � | j � � � � � � | _ d S )Nr )�_loop�_sinkr � create_taskr �_flush�_task)r �loop�sinks r" � create_sourcezAccumulate.create_source. sf � � � ��� ��� � �|�q� � � �D��!�!�"L�"?�/�$�,�"?�"?�� �"L�"L�"N�"N�O�O� � � � r# c � � K � || _ d S )N)r% )r r* s r" � create_sinkzAccumulate.create_sink7 s � � � ��� � � r# c � � K � t j | � � � | j � � � d {V �� d S # t j $ r� t � d| j � � | j �a| j � � � t j t j � � 5 | j � d {V �� d d d � � Y d S # 1 swxY w Y Y d S Y d S w xY w)Nz5Timeout (%ss) sending messages to server on shutdown.) �asyncio�wait_for�stopr � TimeoutError�logger�errorr) �cancel� contextlib�suppress�CancelledError�r s r" �shutdownzAccumulate.shutdown: sC � � � � %��"�4�9�9�;�;��0F�G�G� G� G� G� G� G� G� G� G� G���#� %� %� %� �L�L�G��&� � � � �z�%�� �!�!�#�#�#� �(��)?�@�@� %� %� �*�$�$�$�$�$�$�$�%� %� %� %� %� %� %� %� %� %� %� %� %���� %� %� %� %� %� %� %� &�%�%� %���s4 �28 �A-C�%C�3 C�C �C�C � C�Cc � �l K � t � d� � | j �\| j � � � t j t j � � 5 | j � d {V �� d d d � � n # 1 swxY w Y t � d� � | � � � � d {V �� d S )Nz Accumulate.stop cancel _taskzAccumulate.stop wait lock) r4 �infor) r6 r7 r8 r0 r9 r( r: s r" r2 zAccumulate.stopH s� � � � �� � �2�3�3�3� �:� !� �J� � � � � ��$�W�%;�<�<� !� !��j� � � � � � � � !� !� !� !� !� !� !� !� !� !� !���� !� !� !� !�� � �/�0�0�0��k�k�m�m���������s �A5�5A9�<A9�messagec � �D K � t |j t � � r|j n|j f}|� � � r`t j � d� � 5 |D ]"}| j | � |� � �# d d d � � d S # 1 swxY w Y d S d S )N� accumulate) � isinstance� LIST_CLASS�tuple� do_accumulater �track�taskr �append)r r> � list_types� list_types r" �collectzAccumulate.collectR s � � � � �'�,�e�4�4� '�G� � ��$�&� � � � � "� "� :��!�&�&�|�4�4� :� :�!+� :� :�I��J�y�)�0�0��9�9�9�9�:� :� :� :� :� :� :� :� :� :� :� :� :���� :� :� :� :� :� :� :� :s � &B�B�Bc � � K � | j }t j t � � | _ |� � � D ]�\ }}t |t � � r|� |� � n|f}|D ]�}t � d|j � dt |� � � d�� � | j � ||�� � � � � d {V �� �a# t $ r t � d||� � � w xY w��d S )NzPrepare z(<items=z>) for further processing)�itemsz%s, %s)r r r r rL � issubclassr �batchedr4 r= �__name__�lenr& �process_message� TypeErrorr5 )r � copy_datarI �messagesrN �batchs r" r( zAccumulate._flush^ sB � � � ��J� � �,�T�2�2�� �#,�?�?�#4�#4� � � �I�x� �i��4�4�!� �!�!�(�+�+�+� �[� � !� � ��� � �-�y� 1� -� -�3�u�:�:� -� -� -�� � �� �*�4�4�Y�Y�U�5K�5K�5K�L�L�L�L�L�L�L�L�L�L�� � � � ��L�L��9�e�<�<�<������ � � s �&*C�(C9)rO � __module__� __qualname__r �ProcessingOrder�POST_PROCESS_MESSAGE�PROCESSING_ORDER�SHUTDOWN_PRIORITY�int�os�environ�get�DEFAULT_AGGREGATE_TIMEOUT�SHUTDOWN_SEND_TIMEOUTr r, r. r; r2 r r r rJ r( � __classcell__)r! s @r" r r s) �� � � � � �"�2�G�� � � !$�� � ���>��C�C�!� !� � �C� � ���C�R�H�H� � �� )�.� 3� 3� 3� 3� 3� 3� � � �� � � %� %� %� � � � �V�K� %�&�&� :�]� :� :� :� '�&� :�� � � � � � r# r )r0 r r7 r] �loggingr �defence360agent.apir �"defence360agent.contracts.messagesr r r �!defence360agent.contracts.pluginsr r r �defence360agent.utilsr rO r4 r r r# r" �<module>rh s �� ���� � � � � � � � � � � � � � � � � � � *� *� *� *� *� *�� � � � � � � � � � � � � � � � � � � � 2� 1� 1� 1� 1� 1� ��8� � ��^� ^� ^� ^� ^��m� ^� ^� ^� ^� ^r#
geri dön