o /iF"@sdZddlZddlZddlmZmZmZddlmZddl m Z ddl m Z ddl mZeeZGdd d ZGd d d e ZdS) z2Asynchronous bi-directional streaming RPC helpers.N)CallableOptionalUnion)aio) exceptions) BidiRpcBase)Messagec @sVeZdZdZ d dejdeeee geffddfddZ de fdd Z d d Z dS) _AsyncRequestQueueGeneratora_AsyncRequestQueueGenerator is a helper class for sending asynchronous requests to a gRPC stream from a Queue. This generator takes asynchronous requests off a given `asyncio.Queue` and yields them to gRPC. It's useful when you have an indeterminate, indefinite, or otherwise open-ended set of requests to send through a request-streaming (or bidirectional) RPC. Example:: requests = _AsyncRequestQueueGenerator(q) call = await stub.StreamingRequest(requests) requests.call = call async for response in call: print(response) await q.put(...) Args: queue (asyncio.Queue): The request queue. initial_request (Union[ProtobufMessage, Callable[[], ProtobufMessage]]): The initial request to yield. This is done independently of the request queue to allow for easily restarting streams that require some initial configuration request. Nqueueinitial_requestreturncCs||_||_d|_dSN)_queue_initial_requestcall)selfr r rN/var/www/passon-env/lib/python3.10/site-packages/google/api_core/bidi_async.py__init__>s z$_AsyncRequestQueueGenerator.__init__cCs|jdup |j S)z5Returns true if the call is not set or not completed.Nrdonerrrr _is_activeIsz&_AsyncRequestQueueGenerator._is_activecCs|jdurt|jr|Vn|jV |jIdH}|dur)tddS|s=|j|IdHtddS|Vq)NTz"Cleanly exiting request generator.zEInactive call, replacing item on queue and exiting request generator.)rcallablerget_LOGGERdebugrput)ritemrrr __aiter__Qs$    z%_AsyncRequestQueueGenerator.__aiter__r )__name__ __module__ __qualname____doc__asyncioQueuerrProtobufMessagerrboolrrrrrrr s"  r c@sfeZdZdZdejfddZdddZddd Zd e ddfd d Z de fd dZ e de fddZdS) AsyncBidiRpca:A helper for consuming a async bi-directional streaming RPC. This maps gRPC's built-in interface which uses a request iterator and a response iterator into a socket-like :func:`send` and :func:`recv`. This is a more useful pattern for long-running or asymmetric streams (streams where there is not a direct correlation between the requests and responses). Example:: initial_request = example_pb2.StreamingRpcRequest( setting='example') rpc = AsyncBidiRpc( stub.StreamingRpc, initial_request=initial_request, metadata=[('name', 'value')] ) await rpc.open() while rpc.is_active: print(await rpc.recv()) await rpc.send(example_pb2.StreamingRpcRequest( data='example')) await rpc.close() This does *not* retry the stream on errors. Args: start_rpc (grpc.aio.StreamStreamMultiCallable): The gRPC method used to start the RPC. initial_request (Union[ProtobufMessage, Callable[[], ProtobufMessage]]): The initial request to yield. This is useful if an initial request is needed to start the stream. metadata (Sequence[Tuple(str, str)]): RPC metadata to include in the request. r cCstS)zCreate a queue for requests.)r$r%rrrr _create_queueszAsyncBidiRpc._create_queueNc s|jrtdt|j|jd}z |j||jdIdH}Wntjy1}z| |j d}~ww||_ t |drB|j |j n||j ||_||_ dS)zOpens the stream.z#Cannot open an already open stream.)r )metadataN_wrapped) is_active ValueErrorr _request_queuer _start_rpc _rpc_metadatarGoogleAPICallError _on_call_doneresponserhasattrr+add_done_callback_request_generator)rrequest_generatorrexcrrropens&    zAsyncBidiRpc.opencs>|jdur |j|jdIdHd|_d|_g|_dS)zCloses the stream.N)rcancelr.rr6r _callbacksrrrrcloses   zAsyncBidiRpc.closerequestcsH|jdur td|js|j|IdHdS|jIdHdS)zQueue a message to be sent on the stream. If the underlying RPC has been closed, this will raise. Args: request (ProtobufMessage): The request to send. Nz8Cannot send on an RPC stream that has never been opened.)rr-rr.rread)rr=rrrsends   zAsyncBidiRpc.sendcs$|jdur td|jIdHS)zWait for a message to be returned from the stream. If the underlying RPC has been closed, this will raise. Returns: ProtobufMessage: The received message. Nz8Cannot recv on an RPC stream that has never been opened.)rr-r>rrrrrecvs zAsyncBidiRpc.recvcCs|jduo |j S)z0Whether the stream is currently open and active.Nrrrrrr,szAsyncBidiRpc.is_active)r N)r r!r"r#r$r%r)r9r<r&r?r@propertyr'r,rrrrr(|s(    r()r#r$loggingtypingrrrgrpcrgoogle.api_corergoogle.api_core.bidi_basergoogle.protobuf.messagerr& getLoggerr rr r(rrrrs     \