
    9h                     Z    S SK r S SKrS SKrS SKrS SKrS SKrS SKrS SKr " S S\5      r	g)    Nc                   x   ^  \ rS rSrSrU 4S jrS r\S 5       rS r	SS jr
S r\S	 5       r\S
 5       rSrU =r$ )VerifiableClient   zo
Generic base class for a kafkatest verifiable client.
Implements the common kafkatest protocol and semantics.
c                 :  > [         [        U ]  5         Xl        S[        R
                  " 5       -   U R                  S'   SU l        [        R                  " [        R                  U R                  5        U R                  S[        R                  " 5       -  5        g)z	
        zpython@z	client.idTz	Pid is %dN)superr   __init__confsocketgethostnamerunsignalSIGTERMsig_termdbgosgetpid)selfr	   	__class__s     pC:\Suresh\moveshuttle\MDcreated\moveengine\venv\Lib\site-packages\confluent_kafka/kafkatest/verifiable_client.pyr   VerifiableClient.__init__    sh     	.0	!*V-?-?-A!A		+fnndmm4ryy{*+    c                 4    U R                  S5        SU l        g )Nr   F)r   r   )r   sigframes      r   r   VerifiableClient.sig_term*   s    r   c                  V    [         R                  " S[         R                  " 5       5      $ )Nz%H:%M:%S)timestrftime	localtime r   r   
_timestampVerifiableClient._timestamp.   s    }}Z)9::r   c                 p    [         R                  R                  SU R                  5       < SU< S35        g)zDebugging printout % z DEBUG: 
N)sysstderrwriter!   )r   ss     r   r   VerifiableClient.dbg2   s     

0A1EFr   c                     [         R                  R                  SU R                  5       < SU< S35        U(       a6  [         R                  R                  S5        [         R                  " S5        gg)zEError printout, if term=True the process will terminate immediately. r$   z ERROR: r%   z%% FATAL ERROR ^
   N)r&   r'   r(   r!   exit)r   r)   terms      r   errVerifiableClient.err6   sF    

0A1EFJJ12HHQK r   c                 P   [        [        R                  R                  5       5      US'   U R                  S[        R
                  " U5      -  5        [        R                  R                  S[        R
                  " U5      -  5        [        R                  R                  5         g)z@Send dict as JSON to stdout for consumtion by kafkatest handler _timezSEND: %sz%s
N)
strdatetimenowr   jsondumpsr&   stdoutr(   flush)r   ds     r   sendVerifiableClient.send=   sg    **..01'
djjm+,

$**Q-/0

r   c                    UR                  5        Hq  u  p#Uc  M
  UR                  S5      (       a	  X0USS '   M)  UR                  S5      (       d  MA  USS nUS:X  a!  [        R                  " SS U5      nUS	:X  a  S
nX0U'   Ms     g)z.Set client config properties using args dict. N
topicconf_
   conf_   zpartition.assignment.strategyz/org.apache.kafka.clients.consumer.(\w+)Assignorc                 @    U R                  S5      R                  5       $ )Nr,   )grouplower)xs    r   <lambda>-VerifiableClient.set_config.<locals>.<lambda>[   s    QWWQZ%5%5%7r   stickyzcooperative-sticky)	iteritems
startswithresub)r	   argsnvs       r   
set_configVerifiableClient.set_configD   s     NN$DAy||L)) QrsV<<(( !"A 33 FFM7<=,AG1 %r   c                 J   0 n[        U S5       nU Hs  nUR                  5       nUR                  S5      (       d  [        U5      S:X  a  M:  UR	                  S5      nUS:  a  [        U < SU< 35      eUSU nX4S-   S nXaU'   Mu     SSS5        U$ ! , (       d  f       U$ = f)z>Read (java client) config file and return dict with propertiesr#r   =r,   z#: invalid line, no key=value pair: N)openstriprJ   lenfind	Exception)pathr	   flinefikrO   s          r   read_config_file!VerifiableClient.read_config_filea   s     $_zz|??3''3t9>YYs^6#QUW[$\]]"IAKQ    ! _  s   A:B
B")r	   r   )F)__name__
__module____qualname____firstlineno____doc__r   r   staticmethodr!   r   r/   r;   rP   r`   __static_attributes____classcell__)r   s   @r   r   r      s^    
, ; ;G  8  r   r   )
r4   r6   r   rK   r   r
   r&   r   objectr   r    r   r   <module>rk      s-       	 	   
 \v \r   