src/developer/ffx/lib/fuchsia-controller/python/fidl/_server.py - fuchsia - Git at Google

 # Copyright 2023 The Fuchsia Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.

 import inspect
 import logging
 from abc import abstractmethod
 from inspect import getframeinfo, stack
 from typing import Any, cast

 import fuchsia_controller_py as fc
 from fidl_codec import decode_fidl_request, encode_fidl_message

 from ._fidl_common import (
     DomainError,
     FidlMessage,
     FidlMeta,
     FrameworkError,
     GenericResult,
     parse_ordinal,
     parse_txid,
 )
 from ._ipc import GlobalHandleWaker, HandleWaker

 # Rather than make a long server UUID, this will be a monotonically increasing
 # ID to differentiate servers for debugging purposes.
 _SERVER_ID = 0
 _LOGGER = logging.getLogger("fidl.server")


 class ServerError(Exception):
     pass


 class ServerBase(
     metaclass=FidlMeta,
     required_class_variables=[
         ("library", str),
         ("method_map", dict),
     ],
 ):
     """Base object for doing basic FIDL server tasks."""

     _channel: fc.Channel | None
     library: str
     method_map: dict[int, Any]

     @staticmethod
     @abstractmethod
     def construct_response_object(
         response_ident: str, response_obj: Any
     ) -> Any:
         ...

     def __str__(self) -> str:
         return f"server:{type(self).__name__}:{id(self)}"

     def __init__(
         self,
         channel: fc.Channel,
         channel_waker: HandleWaker | None = None,
     ) -> None:
         global _SERVER_ID
         self._channel = channel
         self.id = _SERVER_ID
         _SERVER_ID += 1
         if channel_waker is None:
             self._channel_waker: HandleWaker = GlobalHandleWaker()
         else:
             self._channel_waker = channel_waker
         caller = getframeinfo(stack()[1][0])
         _LOGGER.debug(
             f"{self} instantiated from {caller.filename}:{caller.lineno}"
         )

     async def serve(self) -> None:
         if self._channel is None:
             raise ValueError("Channel is already closed")

         try:
             with self._channel_waker.registration(
                 self._channel, name=str(self)
             ):
                 while await self.handle_next_request():
                     pass
         finally:
             # As long as we don't do something silly like catch a
             # `GeneratorExit` exception or a `CancelledError` in the above
             # code we should always execute this `finally` block.
             _LOGGER.debug(f"{self} completed serving. Closing channel")
             # Explicitly close the channel instead of deferring closure to the
             # garbage collector to close it. The garbage collector may never
             # close the channel since removing the last reference to an object
             # only ever makes it eligible for garbage collection. One cannot
             # treat these objects like scoped RAII objects in C++ or Rust.
             #
             # If another coroutine depends on this server functioning (like a
             # client), then it'll hang forever. So, we must close the channel in
             # order to make progress.
             if self._channel is not None:
                 self._channel.close()

     async def handle_next_request(self) -> bool:
         # TODO(b/299946378): Handle case where ordinal is unknown.
         # TODO(b/303532690): When attempting to decode a method that is
         # unrecognized, there should be a message sent declaring this is
         # an unknown method.
         try:
             msg, txid, ordinal = await self._channel_read_and_parse()
         except fc.FcTransportStatus as e:
             if e.code() == fc.FcTransportStatus.FC_ERR_FDOMAIN:
                 _LOGGER.debug(f"{self} shutting down. PEER_CLOSED received")
                 return False
             else:
                 _LOGGER.warn(f"{self} channel received error: {e}")
                 raise e
         info = self.method_map[ordinal]
         info.request_ident
         method_name = info.name
         method = getattr(self, method_name)
         if msg is not None:
             res = method(msg)
         else:
             res = method()
         if inspect.isawaitable(res):
             res = await res
         if res is not None and not info.requires_response:
             raise ServerError(
                 f"{self} method {info.name} received a "
                 + "response but is one-way method"
             )
         if res is None and info.requires_response and not info.empty_response:
             raise ServerError(
                 f"{self} method {info.name} returned "
                 + "None when a response was expected"
             )
         if info.has_result:
             _LOGGER.debug(f"{self} received method response {res}")
             if type(res) is DomainError:
                 res = GenericResult(
                     fidl_type=info.response_identifier, err=res.error
                 )
             elif type(res) is FrameworkError:
                 res = GenericResult(
                     fidl_type=info.response_identifier, framework_err=res
                 )
             else:
                 if res is None:
                     res = GenericResult(
                         fidl_type=info.response_identifier, response=object()
                     )
                 else:
                     res = GenericResult(
                         fidl_type=info.response_identifier, response=res
                     )
         if res is not None:
             encoded_fidl_message = encode_fidl_message(
                 ordinal=ordinal,
                 object=res,
                 library=self.library,
                 txid=txid,
                 type_name=res.__fidl_raw_type__,
             )
             if self._channel is None:
                 raise ValueError("Channel is already closed")
             self._channel.write(encoded_fidl_message)
         elif info.empty_response:
             encoded_fidl_message = encode_fidl_message(
                 ordinal=ordinal,
                 object=None,
                 library=self.library,
                 txid=txid,
                 type_name=None,
             )
             if self._channel is None:
                 raise ValueError("Channel is already closed")
             self._channel.write(encoded_fidl_message)
         return True

     async def _channel_read(self) -> FidlMessage:
         while True:
             if self._channel is None:
                 raise ValueError("Channel is already closed")
             try:
                 return self._channel.read()
             except fc.FcTransportStatus as e:
                 # Any number of spurious wakeups are possible. Stay in the loop if the error
                 # is FC_ERR_SHOULD_WAIT.
                 if e.code() == fc.FcTransportStatus.FC_ERR_SHOULD_WAIT:
                     _LOGGER.debug(f"{self} channel spurious wakeup")
                     await self._channel_waker.wait_ready(self._channel)
                     continue
                 _LOGGER.warning(f"{self} channel received error: {e}")
                 raise e

     async def _channel_read_and_parse(self) -> tuple[Any, int, int]:
         raw_msg = await self._channel_read()
         ordinal = parse_ordinal(raw_msg)
         txid = parse_txid(raw_msg)
         handles = raw_msg[1]
         verified_handles: list[int] = [0] * len(handles)
         for i in range(len(handles)):
             # Asserting is not enough for mypy, we must also cast.
             hdl = cast(fc.BaseHandle, handles[i])
             assert isinstance(hdl, fc.BaseHandle)
             verified_handles[i] = hdl.take()
         msg = decode_fidl_request(bytes=raw_msg[0], handles=verified_handles)
         result_obj = self.construct_response_object(
             self.method_map[ordinal].request_ident, msg
         )
         return result_obj, txid, ordinal

     def _send_event(self, ordinal: int, library: str, msg_obj: Any) -> None:
         type_name = None
         if msg_obj is not None:
             type_name = msg_obj.__fidl_raw_type__
         encoded_fidl_message = encode_fidl_message(
             ordinal=ordinal,
             object=msg_obj,
             library=library,
             txid=0,
             type_name=type_name,
         )
         if self._channel is None:
             raise ValueError("Channel is already closed")
         self._channel.write(encoded_fidl_message)
	# Copyright 2023 The Fuchsia Authors. All rights reserved.
	# Use of this source code is governed by a BSD-style license that can be
	# found in the LICENSE file.

	import inspect
	import logging
	from abc import abstractmethod
	from inspect import getframeinfo, stack
	from typing import Any, cast

	import fuchsia_controller_py as fc
	from fidl_codec import decode_fidl_request, encode_fidl_message

	from ._fidl_common import (
	DomainError,
	FidlMessage,
	FidlMeta,
	FrameworkError,
	GenericResult,
	parse_ordinal,
	parse_txid,
	)
	from ._ipc import GlobalHandleWaker, HandleWaker

	# Rather than make a long server UUID, this will be a monotonically increasing
	# ID to differentiate servers for debugging purposes.
	_SERVER_ID = 0
	_LOGGER = logging.getLogger("fidl.server")


	class ServerError(Exception):
	pass


	class ServerBase(
	metaclass=FidlMeta,
	required_class_variables=[
	("library", str),
	("method_map", dict),
	],
	):
	"""Base object for doing basic FIDL server tasks."""

	_channel: fc.Channel \| None
	library: str
	method_map: dict[int, Any]

	@staticmethod
	@abstractmethod
	def construct_response_object(
	response_ident: str, response_obj: Any
	) -> Any:
	...

	def __str__(self) -> str:
	return f"server:{type(self).__name__}:{id(self)}"

	def __init__(
	self,
	channel: fc.Channel,
	channel_waker: HandleWaker \| None = None,
	) -> None:
	global _SERVER_ID
	self._channel = channel
	self.id = _SERVER_ID
	_SERVER_ID += 1
	if channel_waker is None:
	self._channel_waker: HandleWaker = GlobalHandleWaker()
	else:
	self._channel_waker = channel_waker
	caller = getframeinfo(stack()[1][0])
	_LOGGER.debug(
	f"{self} instantiated from {caller.filename}:{caller.lineno}"
	)

	async def serve(self) -> None:
	if self._channel is None:
	raise ValueError("Channel is already closed")

	try:
	with self._channel_waker.registration(
	self._channel, name=str(self)
	):
	while await self.handle_next_request():
	pass
	finally:
	# As long as we don't do something silly like catch a
	# `GeneratorExit` exception or a `CancelledError` in the above
	# code we should always execute this `finally` block.
	_LOGGER.debug(f"{self} completed serving. Closing channel")
	# Explicitly close the channel instead of deferring closure to the
	# garbage collector to close it. The garbage collector may never
	# close the channel since removing the last reference to an object
	# only ever makes it eligible for garbage collection. One cannot
	# treat these objects like scoped RAII objects in C++ or Rust.
	#
	# If another coroutine depends on this server functioning (like a
	# client), then it'll hang forever. So, we must close the channel in
	# order to make progress.
	if self._channel is not None:
	self._channel.close()

	async def handle_next_request(self) -> bool:
	# TODO(b/299946378): Handle case where ordinal is unknown.
	# TODO(b/303532690): When attempting to decode a method that is
	# unrecognized, there should be a message sent declaring this is
	# an unknown method.
	try:
	msg, txid, ordinal = await self._channel_read_and_parse()
	except fc.FcTransportStatus as e:
	if e.code() == fc.FcTransportStatus.FC_ERR_FDOMAIN:
	_LOGGER.debug(f"{self} shutting down. PEER_CLOSED received")
	return False
	else:
	_LOGGER.warn(f"{self} channel received error: {e}")
	raise e
	info = self.method_map[ordinal]
	info.request_ident
	method_name = info.name
	method = getattr(self, method_name)
	if msg is not None:
	res = method(msg)
	else:
	res = method()
	if inspect.isawaitable(res):
	res = await res
	if res is not None and not info.requires_response:
	raise ServerError(
	f"{self} method {info.name} received a "
	+ "response but is one-way method"
	)
	if res is None and info.requires_response and not info.empty_response:
	raise ServerError(
	f"{self} method {info.name} returned "
	+ "None when a response was expected"
	)
	if info.has_result:
	_LOGGER.debug(f"{self} received method response {res}")
	if type(res) is DomainError:
	res = GenericResult(
	fidl_type=info.response_identifier, err=res.error
	)
	elif type(res) is FrameworkError:
	res = GenericResult(
	fidl_type=info.response_identifier, framework_err=res
	)
	else:
	if res is None:
	res = GenericResult(
	fidl_type=info.response_identifier, response=object()
	)
	else:
	res = GenericResult(
	fidl_type=info.response_identifier, response=res
	)
	if res is not None:
	encoded_fidl_message = encode_fidl_message(
	ordinal=ordinal,
	object=res,
	library=self.library,
	txid=txid,
	type_name=res.__fidl_raw_type__,
	)
	if self._channel is None:
	raise ValueError("Channel is already closed")
	self._channel.write(encoded_fidl_message)
	elif info.empty_response:
	encoded_fidl_message = encode_fidl_message(
	ordinal=ordinal,
	object=None,
	library=self.library,
	txid=txid,
	type_name=None,
	)
	if self._channel is None:
	raise ValueError("Channel is already closed")
	self._channel.write(encoded_fidl_message)
	return True

	async def _channel_read(self) -> FidlMessage:
	while True:
	if self._channel is None:
	raise ValueError("Channel is already closed")
	try:
	return self._channel.read()
	except fc.FcTransportStatus as e:
	# Any number of spurious wakeups are possible. Stay in the loop if the error
	# is FC_ERR_SHOULD_WAIT.
	if e.code() == fc.FcTransportStatus.FC_ERR_SHOULD_WAIT:
	_LOGGER.debug(f"{self} channel spurious wakeup")
	await self._channel_waker.wait_ready(self._channel)
	continue
	_LOGGER.warning(f"{self} channel received error: {e}")
	raise e

	async def _channel_read_and_parse(self) -> tuple[Any, int, int]:
	raw_msg = await self._channel_read()
	ordinal = parse_ordinal(raw_msg)
	txid = parse_txid(raw_msg)
	handles = raw_msg[1]
	verified_handles: list[int] = [0] * len(handles)
	for i in range(len(handles)):
	# Asserting is not enough for mypy, we must also cast.
	hdl = cast(fc.BaseHandle, handles[i])
	assert isinstance(hdl, fc.BaseHandle)
	verified_handles[i] = hdl.take()
	msg = decode_fidl_request(bytes=raw_msg[0], handles=verified_handles)
	result_obj = self.construct_response_object(
	self.method_map[ordinal].request_ident, msg
	)
	return result_obj, txid, ordinal

	def _send_event(self, ordinal: int, library: str, msg_obj: Any) -> None:
	type_name = None
	if msg_obj is not None:
	type_name = msg_obj.__fidl_raw_type__
	encoded_fidl_message = encode_fidl_message(
	ordinal=ordinal,
	object=msg_obj,
	library=library,
	txid=0,
	type_name=type_name,
	)
	if self._channel is None:
	raise ValueError("Channel is already closed")
	self._channel.write(encoded_fidl_message)