o
    ci                     @   s   d dl Z d dlZd dlZd dlZd dlmZmZ d dlmZ d dl	m
Z
mZmZmZ d dlZd dlmZ d dlmZmZ d dlmZ G dd	 d	eZG d
d deZdS )    N)ABCabstractmethod)wraps)Callable	CoroutineOptionalUnion)RequestMetadata)calculate_remaining_timeoutgenerate_request_id)RequestCancelledErrorc                   @   s   e Zd Zedee fddZedd Zedd Zedd	 Z	ed
e
fddZedd Zedee dejfddZedejfddZedejfddZdS )ReplicaResult	timeout_sc                 C      t NNotImplementedErrorselfr    r   U/home/ubuntu/.local/lib/python3.10/site-packages/ray/serve/_private/replica_result.pyget      zReplicaResult.getc                       t r   r   r   r   r   r   	get_async      zReplicaResult.get_asyncc                 C   r   r   r   r   r   r   r   __next__   r   zReplicaResult.__next__c                    r   r   r   r   r   r   r   	__anext__   r   zReplicaResult.__anext__callbackc                 C   r   r   r   r   r   r   r   r   add_done_callback    r   zReplicaResult.add_done_callbackc                 C   r   r   r   r   r   r   r   cancel$   r   zReplicaResult.cancelreturnc                 C   r   r   r   r   r   r   r   to_object_ref(   r   zReplicaResult.to_object_refc                    r   r   r   r   r   r   r   to_object_ref_async,   r   z!ReplicaResult.to_object_ref_asyncc                 C   r   r   r   r   r   r   r   to_object_ref_gen0   s   zReplicaResult.to_object_ref_genN)__name__
__module____qualname__r   r   floatr   r   r   r   r   r!   r"   ray	ObjectRefr$   r%   ObjectRefGeneratorr&   r   r   r   r   r      s&    



r   c                   @   s   e Zd Zdeejejf defddZe	de
jfddZdeeef fd	d
Zedee fddZedd Zedd Zedd ZdefddZdd Zdddee dejfddZdejfddZdejfdd ZdS )!ActorReplicaResultobj_ref_or_genmetadatac                    s   d _ d _|j_|j_t _d _	t
|tjr |_n|_ jr/jd us/J dtjj   jrSt _tjj jj  fdd d S d S )Nz<An ObjectRefGenerator must be passed for streaming requests.c                    s   t jj jjS r   )r+   servecontext_remove_in_flight_request_internal_request_id_response_id)_request_contextr   r   r   <lambda>V   s    z-ActorReplicaResult.__init__.<locals>.<lambda>)_obj_ref_obj_ref_genis_streaming_is_streaming
request_id_request_id	threadingLock_object_ref_or_gen_sync_lock$_lazy_object_ref_or_gen_asyncio_lock
isinstancer+   r-   r1   r2   _get_serve_request_contextcancel_on_parent_request_cancelr   r5   _add_in_flight_requestr4   r!   )r   r/   r0   r   r7   r   __init__8   s.   

zActorReplicaResult.__init__r#   c                 C   s   | j du r
t | _ | j S )zLazy `asyncio.Lock` object.N)rC   asynciorA   r   r   r   r   _object_ref_or_gen_asyncio_lock[   s   

z2ActorReplicaResult._object_ref_or_gen_asyncio_lockfc                    s:   t   fdd}t   fdd}t r|S |S )Nc                    s6   z | g|R i |W S  t jjy   t| jw r   )r+   
exceptionsTaskCancelledErrorr   r?   r   argskwargsrK   r   r   wrapperd   s
   
z5ActorReplicaResult._process_response.<locals>.wrapperc                    s<   z | g|R i |I d H W S  t jjy   t w r   )r+   rL   rM   rI   CancelledErrorrN   rQ   r   r   async_wrapperk   s   z;ActorReplicaResult._process_response.<locals>.async_wrapper)r   inspectiscoroutinefunction)rK   rR   rT   r   rQ   r   _process_responsec   s   
z$ActorReplicaResult._process_responser   c                 C   sB   | j rJ dt }| j|d}t||t d}tj||dS )Nz7get() can only be called on a unary ActorReplicaResult.r   )r   start_time_scurr_time_s)timeout)r=   timer$   r
   r+   r   )r   r   rY   
object_refremaining_timeout_sr   r   r   r   w   s   zActorReplicaResult.getc                    s$   | j rJ d|  I d H I d H S )Nz=get_async() can only be called on a unary ActorReplicaResult.)r=   r%   r   r   r   r   r      s   zActorReplicaResult.get_asyncc                 C   s"   | j sJ d| j }t|S )Nz<next() can only be called on a streaming ActorReplicaResult.)r=   r;   r   r+   r   r   next_obj_refr   r   r   r      s   

zActorReplicaResult.__next__c                    s*   | j sJ d| j I d H }|I d H S )NzA__anext__() can only be called on a streaming ActorReplicaResult.)r=   r;   r   r_   r   r   r   r      s   
zActorReplicaResult.__anext__r   c                 C   s.   | j d ur| j  | d S | j| d S r   )r;   	completed_on_completedr:   r    r   r   r   r!      s   
z$ActorReplicaResult.add_done_callbackc                 C   s*   | j d urt| j  d S t| j d S r   )r;   r+   r"   r:   r   r   r   r   r"      s   
zActorReplicaResult.cancelNrX   c                C   s~   | j rJ d| j* | jd u r+| jj|d}| rtd|| _W d    | jS W d    | jS 1 s7w   Y  | jS )Nz?to_object_ref can only be called on a unary ReplicaActorResult.rX   z!Timed out resolving to ObjectRef.)r=   rB   r:   r;   
_next_syncis_nilTimeoutError)r   r   obj_refr   r   r   r$      s"   


z ActorReplicaResult.to_object_refc              	      sp   | j rJ d| j4 I d H  | jd u r| j I d H | _W d   I d H  | jS 1 I d H s0w   Y  | jS )NzEto_object_ref_async can only be called on a unary ReplicaActorResult.)r=   rJ   r:   r;   r   r   r   r   r   r%      s   
z&ActorReplicaResult.to_object_ref_asyncc                 C   s   | j sJ d| jS )NzGto_object_ref_gen can only be called on a streaming ReplicaActorResult.)r=   r;   r   r   r   r   r&      s
   z$ActorReplicaResult.to_object_ref_gen)r'   r(   r)   r   r+   r,   r-   r	   rH   propertyrI   rA   rJ   r   r   rW   r   r*   r   r   r   r   r!   r"   r$   r%   r&   r   r   r   r   r.   7   s,    
#


r.   )rI   rU   r@   r\   abcr   r   	functoolsr   typingr   r   r   r   r+   ray.serve._private.commonr	   ray.serve._private.utilsr
   r   ray.serve.exceptionsr   r   r.   r   r   r   r   <module>   s    (