U
    ^»{jÓ2  ã                   @  sÀ   d dl mZ d dlZd dlZd dlmZ d dlmZ d dlZd dl	m
Z
mZ d dlmZ er”d dlmZmZ d dlmZ d d	lmZ d d
lmZmZ e e¡ZG dd„ deƒZG dd„ dƒZdS )é    )ÚannotationsN)Úpartial)ÚTYPE_CHECKING)ÚchannelsÚprotocol)Ú_manual_thread_pool)ÚAnyÚMutableMapping)ÚBrowser)ÚChannelInterface)ÚSessionÚTargetc                   @  s   e Zd ZdS )ÚUnhandledMessageWarningN)Ú__name__Ú
__module__Ú__qualname__© r   r   ú\/var/www/html/bwave/python/venv/lib/python3.8/site-packages/choreographer/_brokers/_async.pyr      s   r   c                   @  s¢   e Zd ZU dZded< ded< ded< ded	< ddd
dœdd„Zddddœdd„Zd
dœdd„Zd
dœdd„Zdddœdd„Z	dddœdd „Z
dd!d"œd#d$„Zd%S )&ÚBrokerzGBroker is a middleware implementation for asynchronous implementations.r
   Ú_browserr   Ú_channelz8MutableMapping[protocol.MessageKey, asyncio.Future[Any]]ÚfutureszCMutableMapping[str, MutableMapping[str, list[asyncio.Future[Any]]]]Ú_subscriptions_futuresÚNone)ÚbrowserÚchannelÚreturnc                 C  sL   || _ || _tƒ | _tƒ | _d| _i | _i | _t 	¡ | _
tjddd�| _dS )zÔ
        Construct a broker for a synchronous arragenment w/ both ends.

        Args:
            browser: The sync browser implementation.
            channel: The channel the browser uses to talk on.

        Né   Zreadwrite_thread)Úmax_workersÚname)r   r   ÚsetÚ_background_tasksÚ_background_tasks_cancellableÚ_current_read_taskr   r   ÚasyncioÚLockÚ_write_lockr   ZManualThreadExecutorÚ	_executor)Úselfr   r   r   r   r   Ú__init__4   s    	
þzBroker.__init__Ústrúasyncio.Future[Any])Ú
session_idÚsubscriptionr   c                 C  sl   t  d|› d|› d�¡ || jkr,i | j|< || j| krHg | j| |< t ¡  ¡ }| j| |  |¡ |S )NzSession z is subscribing to z
 one time.)Ú_loggerÚdebugr   r$   Úget_running_loopÚcreate_futureÚappend)r(   r,   r-   Úfuturer   r   r   Únew_subscription_futureM   s    ÿ

zBroker.new_subscription_future©r   c                 C  s  t  d¡ | j ¡ D ]$}| ¡ st  d|› �¡ | ¡  qt  d¡ | jrp| j ¡ spt  d| j› �¡ | j ¡  t  d¡ | j ¡ D ]<}| ¡ D ].}|D ]$}| ¡ s˜t  d|› �¡ | ¡  q˜q�q„t  d¡ | j	D ]$}| ¡ sÒt  d|› �¡ | ¡  qÒ| j
jddd� d S )	NzCancelling message futureszCancelling zCancelling read taskzCancelling read: zCancelling subscription-futureszCancelling background tasksT)ÚwaitZcancel_futures)r.   r/   r   ÚvaluesÚdoneÚdebug2Úcancelr#   r   r"   r'   Úshutdown)r(   r3   ÚsessionÚqueryÚtaskr   r   r   Úclean]   s,    







zBroker.cleanc                   sJ   dddœ‡fdd„‰ ddœ‡ ‡‡fdd„‰t  ˆƒ ¡}| ˆ ¡ |ˆ_d S )	Nr+   r   )Úresultr   c                   s¦   |   ¡ rt d¡ d S |  ¡ }|r¢t d¡ ˆ j t ˆ j 	¡ ¡¡ t
|tjƒrjt d¡ tjd|d� n8t
|tjƒr�t d¡ tjd|d� ntjd|d� |‚d S )NzReadloop cancelledz,Error in readloop. Will post a close() task.zPipeClosedError caughtzFull Error:)Úexc_infozCancelledError caught.zError in run_read_loop.)Ú	cancelledr.   r/   Ú	exceptionr!   Úaddr$   Úcreate_taskr   ÚcloseÚ
isinstancer   ÚChannelClosedErrorr9   ÚCancelledErrorÚerror)r@   Úe)r(   r   r   Úcheck_read_loop_errorv   s"    

ÿ

z3Broker.run_read_loop.<locals>.check_read_loop_errorr5   c               
   “  sÒ  t  ¡ } tˆjjdd�}| jˆj|d�I d H }t dt	|ƒ› d�¡ |D �]f}t
 |¡}t
 |¡}|st|rtt
 |¡‚ˆ |¡ t
 |¡�r | dd¡}t d|› �¡ ˆ |¡}|s¶qH|\}}	|	sÎt d	¡ qHt d
|d › d|› d|	› d�¡ ˆj |¡}
t d¡ |
�r„|
D ]r}| d¡�r8|d  |d d… ¡�pB|d |k}|�rt d¡ |
| D ]}| ¡ �s\| |¡ �q\g |
|< �qt d¡ t|	jƒD ]„}| d¡�rÀ|d  |d d… ¡�pÊ|d |k}t d¡ |�r˜t  |	j| d |ƒ¡}ˆj |¡ |	j| d �s˜|	 |¡ �q˜qH|�ršt d|› �¡ |ˆjk�r`t d|› �¡ ˆj  |¡}n$d|k�rvt
 |¡‚nt!d|› �ƒ‚| ¡ �s°| |¡ qHt"j#d|›�t$dd� qHt  ˆƒ ¡}| %ˆ ¡ |ˆ_&d S )NT)Úblocking)ÚexecutorÚfunczChannel read found z json objects.Ú	sessionIdÚ zIs event for z(Found an event that returned no session.zReceived event Úmethodz for z targeting Ú.z'Checking for event subscription future.Ú*éÿÿÿÿz Found event subscription future.z)Checking for event subscription callback.z"Found event subscription callback.r   é   zHave a response with key zFound future for key rJ   z Couldn't find a future for key: zUnhandled message type:)Ú
stacklevel)'r$   r0   r   r   Z
read_jsonsÚrun_in_executorr'   r.   r/   Úlenr   Zget_error_from_resultÚcalculate_message_keyZDevtoolsProtocolErrorÚ_check_for_closed_sessionZis_eventÚgetr9   Ú!_get_target_session_by_session_idrJ   r   ÚendswithÚ
startswithr8   Z
set_resultÚlistZsubscriptionsrE   r"   rD   Zunsubscriber   ÚpopÚRuntimeErrorÚwarningsÚwarnr   Úadd_done_callbackr#   )ÚloopÚfnÚ	responsesÚresponserJ   ÚkeyZevent_session_idÚxÚ_Zevent_sessionZsession_futuresr=   Úmatchr3   ÚtÚ	read_task©rL   Ú	read_loopr(   r   r   rq   Š   s²    þ




þÿ
ÿÿÿþ
ýÿ
ÿþ
ýÿÿ

ý
z'Broker.run_read_loop.<locals>.read_loop)r$   rE   re   r#   )r(   ro   r   rp   r   Úrun_read_loopu   s
    b
zBroker.run_read_loopzprotocol.BrowserCommandzprotocol.BrowserResponse)Úobjr   c              
   Ã  sT  t  |¡ t  |¡}t d|d › d|› �¡ |s:tdƒ‚t ¡ }| ¡ }|| j	|< t 
d|› d|› �¡ zF| j4 I d H š( t ¡ }| | j| jj|¡I d H  W 5 Q I d H R X W n˜ tjtjfk
�r } z*| ¡ rÜ| ¡ sæ|I d H  t d¡|‚W 5 d }~X Y nH tk
�rH } z(| |¡ | j	|= t 
d|› d	�¡ W 5 d }~X Y nX |I d H S )
NzBroker writing rR   z
 with key zIMessage strangely formatted and choreographer couldn't figure it out why.zCreated future: ú zExecutor is closed.zFuture for z	 deleted.)r   Zverify_paramsrZ   r.   Zdebug1rb   r$   r0   r1   r   r/   r&   rX   r'   r   Ú
write_jsonr   ZExecutorClosedErrorrI   r:   rB   r   rH   Ú	ExceptionZset_exception)r(   rs   rj   rf   r3   rK   r   r   r   ru   ð   s8    

ÿ
ý

$zBroker.write_jsonztuple[Target, Session] | None)r,   r   c                 C  sj   |dkr| j | j j| fS | j j ¡ D ] }||jkr&||j| f  S q&|| j jkrf| j | j j| fS d S )NrQ   )r   ÚsessionsÚtabsr7   )r(   r,   Útabr   r   r   r]     s    
z(Broker._get_target_session_by_session_idÚbool)ri   r   c                 C  s�   d|krˆ|d dkrˆ|d   dd¡}|dkr:t d¡ dS |  |¡}|rR|\}}ndS |r„| |¡ t d	|› d
�¡ t d¡ dS dS dS d S )NrR   zTarget.detachedFromTargetÚparamsrP   rQ   z$Found closed session through events.TFzDUsing intern subscription key: 'Target.detachedFromTarget'. Session z was closed.)r\   r.   r9   r]   Z_remove_sessionr/   )r(   ri   Zsession_closedrk   Ztarget_closedrl   r   r   r   r[      s*    þ




ÿ
z Broker._check_for_closed_sessionN)r   r   r   Ú__doc__Ú__annotations__r)   r4   r?   rr   ru   r]   r[   r   r   r   r   r      s   
	{#r   )Ú
__future__r   r$   rc   Ú	functoolsr   Útypingr   ZlogistroZchoreographerr   r   Zchoreographer.utilsr   r   r	   Zchoreographer.browser_asyncr
   Z&choreographer.channels._interface_typer   Z%choreographer.protocol.devtools_asyncr   r   Ú	getLoggerr   r.   ÚUserWarningr   r   r   r   r   r   Ú<module>   s   
