o
    `jz                     @   s^   d dl Z d dlZd dlZd dlZd dlmZ d dlmZ e	e
ZdddZG dd dZdS )	    N)urlparse c                    s   t |}|||}tdg}|D ]^}	ztd||	| |j| |	||dI dH }
W |
  S  tjt	fyN } z|
td|f  W Y d}~qd}~w tjyZ   td tjyq } z|
| W Y d}~qd}~ww |d )z<Connect to host:port; raises KafkaConnectionError on failurezDNS Resolution failurez-%s: Attempting to connect to %s (options: %s)
timeout_atNzunable to connect: %sConnection timed out)KafkaNetSocket
dns_lookupErrorsKafkaConnectionErrorlogdebugconnectsocketerrorOSErrorappendKafkaTimeoutError)nethostportsocket_options	proxy_urlr   socket_factoryaddrs
exceptionsressocker   r   L/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/net/inet.pycreate_connection   s*   


r    c                       s   e Zd Zi Zedd Z fddZd fdd	Zddd	ZdddZ	e
je
je
jfddZ
dddZdddZdd Z  ZS )r   c                 C   s   |j D ]}|| j|< qd S N)SCHEMES	_registry)clsklassschemer   r   r   register_class'   s   
zKafkaNetSocket.register_classc                    s    t  jdi | t|  d S )Nr   )super__init_subclass__r   r'   )r$   kw	__class__r   r   r)   ,   s   z KafkaNetSocket.__init_subclass__Nc                    s   |d u r
t  | S zt|}W n ty   td|f w |js)td|f ztj|j }W n ty?   td|j w t  |S )NzUnable to parse proxy_url: %sz!proxy_url requires scheme:// (%s)z Unsupported proxy url scheme: %s)	r(   __new__r   	Exception
ValueErrorr&   r   r#   KeyError)r$   r   parsedr%   r+   r   r   r-   0   s   zKafkaNetSocket.__new__c                 C   s   d S r!   r   )selfr   r   r   r   __init__?   s   zKafkaNetSocket.__init__Fc              
   C   sh   zt ||t jt jW S  t jy3 } zd|||f }|s*t| g W  Y d }~S t|d }~ww )NzDNS lookup failed for %s:%d, %r)	r   getaddrinfo	AF_UNSPECSOCK_STREAMgaierrorr   warningr
   r   )r2   r   r   raise_errorexerr_strr   r   r   r	   C   s   

zKafkaNetSocket.dns_lookupc                 C   s   t  |||S r!   )r   )r2   family	sock_typeprotor   r   r   r   R   s   zKafkaNetSocket.socketr   c                    sT   |\}}}}}	|  |||}
|
d |D ]}|
j|  q| j||
|	|dI dH S )zGCreate non-blocking socket (with options) and connect to addrinfo tupleFr   N)r   setblocking
setsockoptsock_connect)r2   r   addrinfor   r   r<   r=   r>   
_canonnamesockaddrr   optionr   r   r   r   U   s   
zKafkaNetSocket.connectc              
      s   |d u st  |k rvd }z| ||}W n! ty!   tj}Y n tjy6 } z	|j}W Y d }~nd }~ww |r>|tjkrFt	
d| |S |tjtjtjdfv r\|j||dI d H  ntj|d}td|||d u st  |k std)NzConnected: %si&'  r   UNKNOWNz{} {}r   )time	monotonic
connect_exBlockingIOErrorerrnoEWOULDBLOCKr   r   EISCONNr   r   EINPROGRESSEALREADY
wait_write	errorcodegetr
   r   formatr   )r2   r   r   rD   r   reterrerrstrr   r   r   rA   ^   s(   

zKafkaNetSocket.sock_connectc                 C   s
   | |S r!   )rI   )r2   r   rD   r   r   r   rI   y   s   
zKafkaNetSocket.connect_exr!   )F)r   N)__name__
__module____qualname__r#   classmethodr'   r)   r-   r3   r	   r   r5   r6   IPPROTO_TCPr   rA   rI   __classcell__r   r   r+   r   r   #   s    




	r   )r   NN)rK   loggingr   rG   urllib.parser   kafka.errorserrorsr
   	getLoggerrW   r   r    r   r   r   r   r   <module>   s    

