o
    `ja                     @   s   d dl Z d dlZd dlZd dlZd dlmZmZ G dd dejZG dd dejZ	dd Z
d	d
 ZedkrAe
  Zee dS dS )    N)KafkaConsumerKafkaProducerc                       $   e Zd Z fddZdd Z  ZS )Producerc                    s*   t    || _|| _|| _d| | _d S )N   1)super__init__bootstrap_serverstopic
stop_eventbig_msgselfr	   r
   r   msg_size	__class__ [/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/benchmarks/load_example.pyr      s
   
zProducer.__init__c                 C   sX   t | jd}d| _| j s"|| j| j |  jd7  _| j r|  |	  d S )N)r	   r      )
r   r	   sentr   is_setsendr
   r   flushclose)r   producerr   r   r   run   s   

zProducer.run__name__
__module____qualname__r   r   __classcell__r   r   r   r   r      s    r   c                       r   )Consumerc                    s&   t    || _|| _|| _|| _d S )N)r   r   r	   r
   r   r   r   r   r   r   r       s
   

zConsumer.__init__c                 C   s   t | jdd}|| jg d| _d| _|D ]*}t|j| jkr(|  jd7  _nt	dt|j| j |  jd7  _| j
 r@ nq|  d S )Nearliest)r	   auto_offset_resetr   r   zInvalid message:)r   r	   	subscriber
   validinvalidlenvaluer   printr   r   r   )r   consumermessager   r   r   r   '   s   
zConsumer.runr   r   r   r   r   r!      s    r!   c                  C   sl   t jdd} | jdtdddd | jdtd	d
d | jdtddd | jdtddd | jdtdd d | S )Nz<This tool is used to demonstrate consumer and producer load.)descriptionz--bootstrap-servers+zlocalhost:9092zAhost:port for cluster bootstrap servers (default: localhost:9092))typenargsdefaulthelpz--topiczBTopic for load test (default: kafka-python-benchmark-load-example)z#kafka-python-benchmark-load-example)r.   r1   r0   z
--msg-sizez7Message size, in bytes, for load test (default: 524288)i   z--load-timez0number of seconds to run load test (default: 10)
   z--log-levelz:Optional logging level for load test: ERROR|INFO|DEBUG etc)argparseArgumentParseradd_argumentstrint)parserr   r   r   get_args_parser:   s8   r9   c                 C   s   | j rtjdtt| j d t }t }t| j| j|| j	t
| j| j|| j	g}|D ]}|  q,t| j |  |  td|d j  td|d j  td|d j  d S )NzO%(asctime)s.%(msecs)s:%(name)s:%(thread)d:%(levelname)s:%(process)d:%(message)s)formatlevelzMessages sent: %dr   zMessages recvd: %dr   zMessages invalid: %d)	log_levelloggingbasicConfiggetattr	threadingEventr   r	   r
   r   r!   starttimesleep	load_timesetr)   r   r%   r&   )argsproducer_stopconsumer_stopthreadstr   r   r   mainT   s$   

rL   __main__)r3   r=   r@   rC   kafkar   r   Threadr   r!   r9   rL   r   
parse_argsrG   r   r   r   r   <module>   s   
