o `šjöã@spddlZddlZddlZddlmZddlmZddlm Z ddl m Z ddl m Z e e¡ZGdd„dƒZdS)éN)ÚFindCoordinatorResponse)Ú KafkaBytes)ÚInt32©Ú __version__c@sZeZdZdZdd„Zdd„Zdd„Zdd d „Zd d „Zd d„Z dd„Z dd„Z dd„Z dS)Ú KafkaProtocola“Manage the kafka network protocol Use an instance of KafkaProtocol to manage bytes send/recv'd from a network socket to a broker. Arguments: client_id (str): identifier string to be included in each request ident (str): Optional log-prefix identifier. receive_message_max_bytes (int): Maximum allowed message frame size. Default: 100000000 (100MB). cKs^| dd¡|_| d| ¡¡|_| dd¡|_d|_tdƒ|_d|_d|_ t   ¡|_ g|_ dS) NÚidentÚÚ client_idÚreceive_message_max_bytesiáõréF)ÚgetÚ_identÚ_gen_client_idÚ _client_idÚ_max_frame_sizeÚ_correlation_idrÚ_headerÚ_rbufferÚ _receivingÚ collectionsÚdequeÚin_flight_requestsÚ bytes_to_send)ÚselfÚconfig©rúS/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/protocol/parser.pyÚ__init__s   zKafkaProtocol.__init__cCs|jdd|_|jS)Nél)r©rrrrÚ_next_correlation_id%sz"KafkaProtocol._next_correlation_idcCsdtS)Nz kafka-pythonrr rrrr)szKafkaProtocol._gen_client_idNcCsf|dur| ¡}t d|j||¡|j||jd|jddd}|j |¡|  ¡r1|j  |j ¡|S)aqEncode and queue a kafka api request for sending. Arguments: request : An un-encoded kafka request. correlation_id (int, optional): Optionally specify an ID to correlate requests with responses. If not provided, an ID will be generated automatically. Returns: correlation_id Nz%s Sending request %d %s)Úcorrelation_idr T)ÚframedÚheader) r!ÚlogÚdebugrÚ with_headerrÚencoderÚappendÚexpect_responserr$)rÚrequestr"ÚdatarrrÚ send_request,s  zKafkaProtocol.send_requestcCsVt|jƒ}|dkr dS|dkr|jd}g|_n d |j¡}g|_t d|j|¡|S)z1Retrieve all pending bytes to send on the networkrórz %s Send: %r)ÚlenrÚjoinr%r&r)rÚnr,rrrÚ send_bytesCs   zKafkaProtocol.send_bytesc Csvd}t|ƒ}g}|rt d|j|¡||kr¹|jsbtd|j ¡||ƒ}|j ||||…¡||7}|j ¡dkrV|j  d¡t   |j¡}|  |¡t |ƒ|_d|_n |j ¡dkrbt d¡‚|jrµt|jƒ}|j ¡}t||||ƒ}|j ||||…¡||7}|j ¡}||kr–t d¡‚||kr |Sd|_|j  d¡| |j¡} | | ¡| ¡||ks|S)a"Process bytes received from the network. Arguments: data (bytes): any length bytes received from a network connection to a kafka broker. Returns: responses (list of (correlation_id, response)): any/all completed responses, decoded from bytes to python objects. Raises: KafkaProtocolError: if the bytes received could not be decoded. CorrelationIdError: if the response does not match the request correlation id. rz %s Recv: %rr Tz+this should not happen - are you threading?z,Receive buffer has more bytes than expected?F)r/r%r&rrÚminrÚtellÚwriteÚseekrÚdecodeÚ_validate_frame_sizerrÚErrorsÚ KafkaErrorÚ_process_responser)Ú _reset_buffer) rr,Úir1Ú responsesÚ bytes_to_readÚnbytesÚ total_bytesÚ staged_bytesÚresprrrÚ receive_bytesSsJ         û   Ü%zKafkaProtocol.receive_bytescCs$|dks ||jkrt d|¡‚dS)NrzInvalid frame length: %d)rr9ÚInvalidReceiveError)rr@rrrr8sÿz"KafkaProtocol._validate_frame_sizec Cs"|jst d¡‚|j ¡}|j}| ¡}|dur,t d|j|j ¡t  d|j|j f¡‚|  |¡}|j}|dkrK|dkrK|t urK|j dkrKt  d¡n ||krXt d||f¡‚z| |¡}Wn$ttjfyƒ| d¡| ¡}t d|||t|ƒ|¡t  d¡‚wt d |j||¡||fS) Nz.No in-flight-request found for server responsez1Unable to find ResponseType for api=%d version=%dz+Unable to find response type for api %d v%drz£Kafka 0.8.2 quirk -- FindCoordinatorResponse Correlation ID does not match request. This should go away once at least one topic has been initialized on the broker.z.Correlation IDs do not match: sent %d, recv %dzUResponse %d [ResponseType: %s RequestHeader: %s]: Unable to decode %d-byte buffer: %rzUnable to decode responsez%s Received response %d %s)rr9ÚCorrelationIdErrorÚpopleftr"Úget_response_classr%ÚerrorÚapi_keyÚ api_versionÚKafkaProtocolErrorÚ parse_headerrÚwarningr7Ú ValueErrorÚstructr6Úreadr/r&r) rÚ read_bufferr$r"Ú response_typeÚresponse_headerÚrecv_correlation_idÚresponseÚbufrrrr;“sD  ÿ  ÿÿ  ý ù zKafkaProtocol._process_responsecCsd|_|j d¡d|_dS)NFr)rrr6rr rrrr<¼s  zKafkaProtocol._reset_buffer)N) Ú__name__Ú __module__Ú __qualname__Ú__doc__rr!rr-r2rDr8r;r<rrrrrs   < )r)rÚloggingrPÚ kafka.errorsÚerrorsr9Úkafka.protocol.metadatarÚkafka.protocol.framerÚ$kafka.protocol.schemas.fields.codecsrÚ kafka.versionrÚ getLoggerrXr%rrrrrÚs