import sys import enum import http import time import asyncio import warnings import ipaddress from types import TracebackType from typing import Generic, Optional, Union, Type, List, Sequence, Any, cast from typing import Dict, Tuple, TYPE_CHECKING try: import ssl as _ssl except ImportError: _ssl = None # type: ignore from h2.config import H2Configuration from multidict import MultiDict from .utils import Wrapper, DeadlineWrapper from .const import Status, Cardinality from .config import Configuration from .stream import send_message, recv_message, StreamIterator from .stream import _RecvType, _SendType from .events import _DispatchChannelEvents from .protocol import H2Protocol, AbstractHandler, Stream as _Stream, Peer from .metadata import Deadline, USER_AGENT, decode_grpc_message, encode_timeout from .metadata import encode_metadata, decode_metadata, _MetadataLike, _Metadata from .metadata import _STATUS_DETAILS_KEY, decode_bin_value from .exceptions import GRPCError, ProtocolError, StreamTerminatedError from .encoding.base import GRPC_CONTENT_TYPE, CodecBase, StatusDetailsCodecBase from .encoding.proto import ProtoCodec, ProtoStatusDetailsCodec from .encoding.proto import _googleapis_available from ._registry import channels as _channels if TYPE_CHECKING: from ._typing import IReleaseStream # noqa _H2_OK = '200' # https://github.com/grpc/grpc/blob/master/doc/http-grpc-status-mapping.md _H2_TO_GRPC_STATUS_MAP = { # 400 str(http.HTTPStatus.BAD_REQUEST.value): Status.INTERNAL, # 401 str(http.HTTPStatus.UNAUTHORIZED.value): Status.UNAUTHENTICATED, # 403 str(http.HTTPStatus.FORBIDDEN.value): Status.PERMISSION_DENIED, # 404 str(http.HTTPStatus.NOT_FOUND.value): Status.UNIMPLEMENTED, # 502 str(http.HTTPStatus.BAD_GATEWAY.value): Status.UNAVAILABLE, # 503 str(http.HTTPStatus.SERVICE_UNAVAILABLE.value): Status.UNAVAILABLE, # 504 str(http.HTTPStatus.GATEWAY_TIMEOUT.value): Status.UNAVAILABLE, # 429 str(http.HTTPStatus.TOO_MANY_REQUESTS.value): Status.UNAVAILABLE, } class Handler(AbstractHandler): connection_lost = False def accept(self, stream: Any, headers: Any, release_stream: Any) -> None: raise NotImplementedError('Client connection can not accept requests') def cancel(self, stream: Any) -> None: pass def close(self) -> None: self.connection_lost = True class Stream(StreamIterator[_RecvType], Generic[_SendType, _RecvType]): """ Represents gRPC method call - HTTP/2 request/stream, and everything you need to communicate with server in order to get response. In order to work directly with stream, you should :py:meth:`ServiceMethod.open` request like this: .. code-block:: python3 request = cafe_pb2.LatteOrder( size=cafe_pb2.SMALL, temperature=70, sugar=3, ) async with client.MakeLatte.open() as stream: await stream.send_message(request, end=True) reply: empty_pb2.Empty = await stream.recv_message() """ # stream state _send_request_done = False _send_message_done = False _end_done = False _recv_initial_metadata_done = False _recv_trailing_metadata_done = False _cancel_done = False _trailers_only: Optional[bool] = None _stream: _Stream _release_stream: 'IReleaseStream' _wrapper_ctx = None #: This property contains initial metadata, received with headers from #: the server. It equals to ``None`` initially, and to a multi-dict object #: after :py:meth:`recv_initial_metadata` coroutine succeeds. initial_metadata: Optional[_Metadata] = None #: This property contains trailing metadata, received with trailers from #: the server. It equals to ``None`` initially, and to a multi-dict object #: after :py:meth:`recv_trailing_metadata` coroutine succeeds. trailing_metadata: Optional[_Metadata] = None #: Connection's peer info of type :py:class:`~grpclib.protocol.Peer` peer: Optional[Peer] = None # stats _messages_sent = 0 _messages_received = 0 def __init__( self, channel: 'Channel', method_name: str, metadata: _Metadata, cardinality: Cardinality, send_type: Type[_SendType], recv_type: Type[_RecvType], *, codec: CodecBase, status_details_codec: Optional[StatusDetailsCodecBase], dispatch: _DispatchChannelEvents, deadline: Optional[Deadline] = None, ) -> None: self._channel = channel self._method_name = method_name self._metadata = metadata self._cardinality = cardinality self._send_type = send_type self._recv_type = recv_type self._codec = codec self._status_details_codec = status_details_codec self._dispatch = dispatch self._deadline = deadline async def send_request(self, *, end: bool = False) -> None: """Coroutine to send request headers with metadata to the server. New HTTP/2 stream will be created during this coroutine call. .. note:: This coroutine will be called implicitly during first :py:meth:`send_message` coroutine call, if not called before explicitly. :param end: end outgoing stream if there are no messages to send in a streaming request """ if self._send_request_done: raise ProtocolError('Request is already sent') if end and not self._cardinality.client_streaming: raise ProtocolError('Unary request requires a message to be sent ' 'before ending outgoing stream') with self._wrapper: protocol = await self._channel.__connect__() stream = protocol.processor.connection\ .create_stream(wrapper=self._wrapper) headers = [ (':method', 'POST'), (':scheme', self._channel._scheme), (':path', self._method_name), (':authority', self._channel._authority), ] if self._deadline is not None: timeout = self._deadline.time_remaining() headers.append(('grpc-timeout', encode_timeout(timeout))) # FIXME: remove this check after this issue gets resolved: # https://github.com/googleapis/googleapis.github.io/issues/27 if self._codec.__content_subtype__ == 'proto': content_type = GRPC_CONTENT_TYPE else: content_type = (GRPC_CONTENT_TYPE + '+' + self._codec.__content_subtype__) headers.extend(( ('te', 'trailers'), ('content-type', content_type), ('user-agent', USER_AGENT), )) metadata, = await self._dispatch.send_request( self._metadata, method_name=self._method_name, deadline=self._deadline, content_type=content_type, ) headers.extend(encode_metadata(metadata)) release_stream = await stream.send_request( headers, end_stream=end, _processor=protocol.processor, ) self._stream = stream self._release_stream = release_stream self.peer = self._stream.connection.get_peer() self._send_request_done = True if end: self._end_done = True async def send_message( self, message: _SendType, *, end: bool = False, ) -> None: """Coroutine to send message to the server. If client sends UNARY request, then you should call this coroutine only once. If client sends STREAM request, then you can call this coroutine as many times as you need. .. warning:: It is important to finally end stream from the client-side when you finished sending messages. You can do this in two ways: - specify ``end=True`` argument while sending last message - and last DATA frame will include END_STREAM flag; - call :py:meth:`end` coroutine after sending last message - and extra HEADERS frame with END_STREAM flag will be sent. First approach is preferred, because it doesn't require sending additional HTTP/2 frame. """ if not self._send_request_done: await self.send_request() end_stream = end if not self._cardinality.client_streaming: if self._send_message_done: raise ProtocolError('Message was already sent') else: end_stream = True if self._end_done: raise ProtocolError('Stream is ended') with self._wrapper: message, = await self._dispatch.send_message(message) await send_message(self._stream, self._codec, message, self._send_type, end=end_stream) self._send_message_done = True self._messages_sent += 1 self._stream.connection.messages_sent += 1 self._stream.connection.last_message_sent = time.monotonic() if end: self._end_done = True async def end(self) -> None: """Coroutine to end stream from the client-side. It should be used to finally end stream from the client-side when we're finished sending messages to the server and stream wasn't closed with last DATA frame. See :py:meth:`send_message` for more details. HTTP/2 stream will have half-closed (local) state after this coroutine call. """ if not self._send_request_done: raise ProtocolError('Request was not sent') if self._end_done: raise ProtocolError('Stream was already ended') if not self._cardinality.client_streaming: if not self._send_message_done: raise ProtocolError('Unary request requires a single message ' 'to be sent') else: # `send_message` must already ended stream self._end_done = True return else: await self._stream.end() self._end_done = True def _raise_for_status(self, headers_map: Dict[str, str]) -> None: status = headers_map[':status'] if status is not None and status != _H2_OK: grpc_status = _H2_TO_GRPC_STATUS_MAP.get(status, Status.UNKNOWN) raise GRPCError(grpc_status, 'Received :status = {!r}'.format(status)) def _raise_for_content_type(self, headers_map: Dict[str, str]) -> None: content_type = headers_map.get('content-type') if content_type is None: raise GRPCError(Status.UNKNOWN, 'Missing content-type header') base_content_type, _, sub_type = content_type.partition('+') sub_type = sub_type or ProtoCodec.__content_subtype__ if ( base_content_type != GRPC_CONTENT_TYPE or sub_type != self._codec.__content_subtype__ ): raise GRPCError(Status.UNKNOWN, 'Invalid content-type: {!r}' .format(content_type)) def _process_grpc_status( self, headers_map: Dict[str, str], ) -> Tuple[Status, Optional[str], Any]: grpc_status = headers_map.get('grpc-status') if grpc_status is None: raise GRPCError(Status.UNKNOWN, 'Missing grpc-status header') try: status = Status(int(grpc_status)) except ValueError: raise GRPCError(Status.UNKNOWN, ('Invalid grpc-status: {!r}' .format(grpc_status))) else: message, details = None, None if status is not Status.OK: message = headers_map.get('grpc-message') if message is not None: message = decode_grpc_message(message) if self._status_details_codec is not None: details_bin = headers_map.get(_STATUS_DETAILS_KEY) if details_bin is not None: details = self._status_details_codec.decode( status, message, decode_bin_value(details_bin.encode('ascii')) ) return status, message, details def _raise_for_grpc_status( self, status: Status, message: Optional[str], details: Any, ) -> None: if status is not Status.OK: raise GRPCError(status, message, details) async def recv_initial_metadata(self) -> None: """Coroutine to wait for headers with initial metadata from the server. .. note:: This coroutine will be called implicitly during first :py:meth:`recv_message` coroutine call, if not called before explicitly. May raise :py:class:`~grpclib.exceptions.GRPCError` if server returned non-:py:attr:`Status.OK ` in trailers-only response. When this coroutine finishes, you can access received initial metadata by using :py:attr:`initial_metadata` attribute. """ if not self._send_request_done: raise ProtocolError('Request was not sent yet') if self._recv_initial_metadata_done: raise ProtocolError('Initial metadata was already received') with self._wrapper: headers = await self._stream.recv_headers() self._recv_initial_metadata_done = True headers_map = dict(headers) self._raise_for_status(headers_map) self._raise_for_content_type(headers_map) if 'grpc-status' in headers_map: # trailers-only response self._trailers_only = True im = cast(_Metadata, MultiDict()) im, = await self._dispatch.recv_initial_metadata(im) self.initial_metadata = im status, message, details = self._process_grpc_status( headers_map, ) tm = decode_metadata(headers) tm, = await self._dispatch.recv_trailing_metadata( tm, status=status, status_message=message, status_details=details, ) self.trailing_metadata = tm self._raise_for_grpc_status(status, message, details) else: im = decode_metadata(headers) im, = await self._dispatch.recv_initial_metadata(im) self.initial_metadata = im async def recv_message(self) -> Optional[_RecvType]: """Coroutine to receive incoming message from the server. If server sends UNARY response, then you can call this coroutine only once. If server sends STREAM response, then you should call this coroutine several times, until it returns None when the server has ended the stream. To simplify you code in this case, :py:class:`Stream` implements async iterations protocol, so you can use it like this: .. code-block:: python3 async for message in stream: do_smth_with(message) or even like this: .. code-block:: python3 messages = [msg async for msg in stream] HTTP/2 has flow control mechanism, so client will acknowledge received DATA frames as a message only after user consumes this coroutine. :returns: message """ if not self._recv_initial_metadata_done: await self.recv_initial_metadata() with self._wrapper: message = await recv_message(self._stream, self._codec, self._recv_type) if message is not None: message, = await self._dispatch.recv_message(message) self._messages_received += 1 self._stream.connection.messages_received += 1 self._stream.connection.last_message_received = time.monotonic() return message else: return None async def recv_trailing_metadata(self) -> None: """Coroutine to wait for trailers with trailing metadata from the server. .. note:: This coroutine will be called implicitly at exit from this call (context manager's exit), if not called before explicitly. May raise :py:class:`~grpclib.exceptions.GRPCError` if server returned non-:py:attr:`Status.OK ` in trailers. When this coroutine finishes, you can access received trailing metadata by using :py:attr:`trailing_metadata` attribute. """ if (not self._end_done # explicit end and not (not self._cardinality.client_streaming # implicit end and self._send_message_done)): raise ProtocolError('Outgoing stream was not ended') if not self._recv_initial_metadata_done: raise ProtocolError('Initial metadata was not received before ' 'waiting for trailing metadata') if self._recv_trailing_metadata_done: raise ProtocolError('Trailing metadata was already received') if self._trailers_only: self._recv_trailing_metadata_done = True else: with self._wrapper: trailers = await self._stream.recv_trailers() self._recv_trailing_metadata_done = True status, message, details = self._process_grpc_status( dict(trailers), ) tm = decode_metadata(trailers) tm, = await self._dispatch.recv_trailing_metadata( tm, status=status, status_message=message, status_details=details, ) self.trailing_metadata = tm self._raise_for_grpc_status(status, message, details) async def cancel(self) -> None: """Coroutine to cancel this request/stream. Client will send RST_STREAM frame to the server, so it will be explicitly informed that there is nothing to expect from the client regarding this request/stream. """ if not self._send_request_done: raise ProtocolError('Request was not sent yet') if self._cancel_done: raise ProtocolError('Stream was already cancelled') with self._wrapper: await self._stream.reset() # TODO: specify error code self._cancel_done = True async def __aenter__(self) -> 'Stream[_SendType, _RecvType]': if self._deadline is None: self._wrapper = Wrapper() else: self._wrapper = DeadlineWrapper() self._wrapper_ctx = self._wrapper.start(self._deadline) self._wrapper_ctx.__enter__() self._channel._calls_started += 1 self._channel._last_call_started = time.monotonic() return self async def _maybe_finish(self) -> None: if ( not self._cancel_done and not self._stream._transport.is_closing() ): if not self._recv_initial_metadata_done: await self.recv_initial_metadata() if not self._recv_trailing_metadata_done: await self.recv_trailing_metadata() def _maybe_raise(self) -> None: if self._stream.headers is not None: self._raise_for_status(dict(self._stream.headers)) if self._stream.trailers is not None: status, message, details = self._process_grpc_status( dict(self._stream.trailers), ) self._raise_for_grpc_status(status, message, details) elif self._stream.headers is not None: headers_map = dict(self._stream.headers) if 'grpc-status' in headers_map: status, message, details = self._process_grpc_status( headers_map, ) self._raise_for_grpc_status(status, message, details) async def __aexit__( self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType], ) -> None: if not self._send_request_done: return try: reraise = False if exc_val is None: try: await self._maybe_finish() except Exception: exc_type, exc_val, exc_tb = sys.exc_info() reraise = True if isinstance(exc_val, StreamTerminatedError): self._maybe_raise() if reraise: assert exc_val is not None raise exc_val finally: if self._stream.closable: self._stream.reset_nowait() self._release_stream() if self._wrapper_ctx is not None: self._wrapper_ctx.__exit__(exc_type, exc_val, exc_tb) if exc_val is None: self._channel._calls_succeeded += 1 else: self._channel._calls_failed += 1 class _ChannelState(enum.IntEnum): IDLE = 1 CONNECTING = 2 READY = 3 TRANSIENT_FAILURE = 4 class Channel: """ Represents a connection to the server, which can be used with generated stub classes to perform gRPC calls. .. code-block:: python3 channel = Channel() client = cafe_grpc.CoffeeMachineStub(channel) ... request = cafe_pb2.LatteOrder( size=cafe_pb2.SMALL, temperature=70, sugar=3, ) reply: empty_pb2.Empty = await client.MakeLatte(request) ... channel.close() """ _protocol = None # stats _calls_started = 0 _calls_succeeded = 0 _calls_failed = 0 _last_call_started: Optional[float] = None def __init__( self, host: Optional[str] = None, port: Optional[int] = None, *, loop: Optional[asyncio.AbstractEventLoop] = None, path: Optional[str] = None, codec: Optional[CodecBase] = None, status_details_codec: Optional[StatusDetailsCodecBase] = None, ssl: Union[ None, bool, "_ssl.SSLContext", "_ssl.DefaultVerifyPaths" ] = None, config: Optional[Configuration] = None, ): """Initialize connection to the server :param host: server host name. :param port: server port number. :param loop: (deprecated) asyncio-compatible event loop :param path: server socket path. If specified, host and port should be omitted (must be None). :param codec: instance of a codec to encode and decode messages, if omitted ``ProtoCodec`` is used by default :param status_details_codec: instance of a status details codec to decode error details in a trailing metadata, if omitted ``ProtoStatusDetailsCodec`` is used by default :param ssl: ``True`` or :py:class:`~python:ssl.SSLContext` object or ``ssl.DefaultVerifyPaths`` object; if ``True``, default SSL context is used. """ if path is not None and (host is not None or port is not None): raise ValueError("The 'path' parameter can not be used with the " "'host' or 'port' parameters.") else: if host is None: host = '127.0.0.1' if port is None: port = 50051 if _ssl is None: if ssl is not None: raise RuntimeError('SSL is not supported') elif ssl is True: ssl = self._get_default_ssl_context() elif isinstance(ssl, _ssl.DefaultVerifyPaths): ssl = self._get_default_ssl_context(verify_paths=ssl) if codec is None: codec = ProtoCodec() if status_details_codec is None and _googleapis_available(): status_details_codec = ProtoStatusDetailsCodec() if loop: warnings.warn("The loop argument is deprecated and scheduled " "for removal in grpclib 0.5", DeprecationWarning, stacklevel=2) self._host = host self._port = port self._loop = loop or asyncio.get_event_loop() self._path = path self._codec = codec self._status_details_codec = status_details_codec self._ssl = ssl or None self._scheme = 'https' if self._ssl else 'http' self._authority = self._get_authority(self._host, self._port) self._h2_config = H2Configuration( client_side=True, header_encoding='ascii', validate_inbound_headers=False, validate_outbound_headers=False, normalize_inbound_headers=False, normalize_outbound_headers=False, ) self._connect_lock = asyncio.Lock() self._state = _ChannelState.IDLE config = Configuration() if config is None else config self._config = config.__for_client__() self.__dispatch__ = _DispatchChannelEvents() _channels.add(self) def __repr__(self) -> str: return ('Channel({!r}, {!r}, ..., path={!r})' .format(self._host, self._port, self._path)) def _protocol_factory(self) -> H2Protocol: return H2Protocol(Handler(), self._config, self._h2_config) async def _create_connection(self) -> H2Protocol: if self._path is not None: _, protocol = await self._loop.create_unix_connection( self._protocol_factory, self._path, ssl=self._ssl, server_hostname=( self._config.ssl_target_name_override if self._ssl is not None else None ), ) else: _, protocol = await self._loop.create_connection( self._protocol_factory, self._host, self._port, ssl=self._ssl, server_hostname=( self._config.ssl_target_name_override if self._ssl is not None else None ), ) return protocol @property def _connected(self) -> bool: return (self._protocol is not None and not self._protocol.handler.connection_lost) async def __connect__(self) -> H2Protocol: if not self._connected: async with self._connect_lock: self._state = _ChannelState.CONNECTING if not self._connected: try: self._protocol = await self._create_connection() except Exception: self._state = _ChannelState.TRANSIENT_FAILURE raise else: self._state = _ChannelState.READY return cast(H2Protocol, self._protocol) # https://python-hyper.org/projects/h2/en/stable/negotiating-http2.html def _get_default_ssl_context( self, *, verify_paths: Optional['_ssl.DefaultVerifyPaths'] = None, ) -> '_ssl.SSLContext': if verify_paths is not None: cafile = verify_paths.cafile capath = verify_paths.capath else: try: import certifi except ImportError: cafile = None else: cafile = certifi.where() capath = None ctx = _ssl.create_default_context( purpose=_ssl.Purpose.SERVER_AUTH, cafile=cafile, capath=capath, ) ctx.minimum_version = _ssl.TLSVersion.TLSv1_2 ctx.set_ciphers('ECDHE+AESGCM:ECDHE+CHACHA20:DHE+AESGCM:DHE+CHACHA20') ctx.set_alpn_protocols(['h2']) return ctx def _get_authority(self, host: str, port: int) -> str: try: ipv6_address = ipaddress.IPv6Address(host) except ipaddress.AddressValueError: pass else: host = f"[{ipv6_address}]" return "{}:{}".format(host, port) def request( self, name: str, cardinality: Cardinality, request_type: Type[_SendType], reply_type: Type[_RecvType], *, timeout: Optional[float] = None, deadline: Optional[Deadline] = None, metadata: Optional[_MetadataLike] = None, ) -> Stream[_SendType, _RecvType]: if timeout is not None and deadline is None: deadline = Deadline.from_timeout(timeout) elif timeout is not None and deadline is not None: deadline = min(Deadline.from_timeout(timeout), deadline) metadata = cast(_Metadata, MultiDict(metadata or ())) return Stream(self, name, metadata, cardinality, request_type, reply_type, codec=self._codec, status_details_codec=self._status_details_codec, dispatch=self.__dispatch__, deadline=deadline) def close(self) -> None: """Closes connection to the server. """ if self._protocol is not None: self._protocol.processor.close() del self._protocol self._state = _ChannelState.IDLE def __del__(self) -> None: if self._protocol is not None: message = 'Unclosed connection: {!r}'.format(self) warnings.warn(message, ResourceWarning) if self._loop.is_closed(): return else: self.close() self._loop.call_exception_handler({'message': message}) async def __aenter__(self) -> 'Channel': return self async def __aexit__( self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType], ) -> None: self.close() class ServiceMethod(Generic[_SendType, _RecvType]): """ Base class for all gRPC method types """ def __init__( self, channel: Channel, name: str, request_type: Type[_SendType], reply_type: Type[_RecvType], ): self.channel = channel self.name = name self.request_type = request_type self.reply_type = reply_type @property def _cardinality(self) -> Cardinality: raise NotImplementedError def open( self, *, timeout: Optional[float] = None, metadata: Optional[_MetadataLike] = None, ) -> Stream[_SendType, _RecvType]: """Creates and returns :py:class:`Stream` object to perform request to the server. Nothing will happen to the current underlying HTTP/2 connection during this method call. It just initializes :py:class:`Stream` object for you. Actual request will be sent only during :py:meth:`Stream.send_request` or :py:meth:`Stream.send_message` coroutine call. :param float timeout: request timeout (seconds) :param metadata: custom request metadata, dict or list of pairs :return: :py:class:`Stream` object """ return self.channel.request(self.name, self._cardinality, self.request_type, self.reply_type, timeout=timeout, metadata=metadata) class UnaryUnaryMethod(ServiceMethod[_SendType, _RecvType]): """ Represents UNARY-UNARY gRPC method type. .. automethod:: __call__ .. automethod:: open """ _cardinality = Cardinality.UNARY_UNARY async def __call__( self, message: _SendType, *, timeout: Optional[float] = None, metadata: Optional[_MetadataLike] = None, ) -> _RecvType: """Coroutine to perform defined call. :param message: message :param float timeout: request timeout (seconds) :param metadata: custom request metadata, dict or list of pairs :return: message """ async with self.open(timeout=timeout, metadata=metadata) as stream: await stream.send_message(message, end=True) reply = await stream.recv_message() assert reply is not None return reply class UnaryStreamMethod(ServiceMethod[_SendType, _RecvType]): """ Represents UNARY-STREAM gRPC method type. .. automethod:: __call__ .. automethod:: open """ _cardinality = Cardinality.UNARY_STREAM async def __call__( self, message: _SendType, *, timeout: Optional[float] = None, metadata: Optional[_MetadataLike] = None, ) -> List[_RecvType]: """Coroutine to perform defined call. :param message: message :param float timeout: request timeout (seconds) :param metadata: custom request metadata, dict or list of pairs :return: sequence of messages """ async with self.open(timeout=timeout, metadata=metadata) as stream: await stream.send_message(message, end=True) return [message async for message in stream] class StreamUnaryMethod(ServiceMethod[_SendType, _RecvType]): """ Represents STREAM-UNARY gRPC method type. .. automethod:: __call__ .. automethod:: open """ _cardinality = Cardinality.STREAM_UNARY async def __call__( self, messages: Sequence[_SendType], *, timeout: Optional[float] = None, metadata: Optional[_MetadataLike] = None, ) -> _RecvType: """Coroutine to perform defined call. :param messages: sequence of messages :param float timeout: request timeout (seconds) :param metadata: custom request metadata, dict or list of pairs :return: message """ async with self.open(timeout=timeout, metadata=metadata) as stream: for message in messages[:-1]: await stream.send_message(message) if messages: await stream.send_message(messages[-1], end=True) else: await stream.send_request(end=True) reply = await stream.recv_message() assert reply is not None return reply class StreamStreamMethod(ServiceMethod[_SendType, _RecvType]): """ Represents STREAM-STREAM gRPC method type. .. automethod:: __call__ .. automethod:: open """ _cardinality = Cardinality.STREAM_STREAM async def __call__( self, messages: Sequence[_SendType], *, timeout: Optional[float] = None, metadata: Optional[_MetadataLike] = None, ) -> List[_RecvType]: """Coroutine to perform defined call. :param messages: sequence of messages :param float timeout: request timeout (seconds) :param metadata: custom request metadata, dict or list of pairs :return: sequence of messages """ async with self.open(timeout=timeout, metadata=metadata) as stream: for message in messages[:-1]: await stream.send_message(message) if messages: await stream.send_message(messages[-1], end=True) else: await stream.send_request(end=True) return [message async for message in stream]