| import json |
| import threading |
| |
| import time |
| from concurrent import futures |
| |
| from acts import logger |
| |
| SOCKET_TIMEOUT = 60 |
| |
| # The Session UID when a UID has not been received yet. |
| UNKNOWN_UID = -1 |
| |
| |
| class Sl4aException(Exception): |
| """The base class for all SL4A exceptions.""" |
| |
| |
| class Sl4aStartError(Sl4aException): |
| """Raised when sl4a is not able to be started.""" |
| |
| |
| class Sl4aApiError(Sl4aException): |
| """Raised when remote API reports an error.""" |
| |
| |
| class Sl4aConnectionError(Sl4aException): |
| """An error raised upon failure to connect to SL4A.""" |
| |
| |
| class Sl4aProtocolError(Sl4aException): |
| """Raised when there an error in exchanging data with server on device.""" |
| NO_RESPONSE_FROM_HANDSHAKE = 'No response from handshake.' |
| NO_RESPONSE_FROM_SERVER = 'No response from server.' |
| MISMATCHED_API_ID = 'Mismatched API id.' |
| |
| |
| class MissingSl4AError(Sl4aException): |
| """An error raised when an Sl4aClient is created without SL4A installed.""" |
| |
| |
| class RpcClient(object): |
| """An RPC client capable of processing multiple RPCs concurrently. |
| |
| Attributes: |
| _free_connections: A list of all idle RpcConnections. |
| _working_connections: A list of all working RpcConnections. |
| _lock: A lock used for accessing critical memory. |
| max_connections: The maximum number of RpcConnections at a time. |
| Increasing or decreasing the number of max connections does NOT |
| modify the thread pool size being used for self.future RPC calls. |
| _log: The logger for this RpcClient. |
| """ |
| """The default value for the maximum amount of connections for a client.""" |
| DEFAULT_MAX_CONNECTION = 15 |
| |
| class AsyncClient(object): |
| """An object that allows RPC calls to be called asynchronously. |
| |
| Attributes: |
| _rpc_client: The RpcClient to use when making calls. |
| _executor: The ThreadPoolExecutor used to keep track of workers |
| """ |
| |
| def __init__(self, rpc_client): |
| self._rpc_client = rpc_client |
| self._executor = futures.ThreadPoolExecutor( |
| max_workers=max(rpc_client.max_connections - 2, 1)) |
| |
| def rpc(self, name, *args, **kwargs): |
| future = self._executor.submit(name, *args, **kwargs) |
| return future |
| |
| def __getattr__(self, name): |
| """Wrapper for python magic to turn method calls into RPC calls.""" |
| |
| def rpc_call(*args, **kwargs): |
| future = self._executor.submit( |
| self._rpc_client.__getattr__(name), *args, **kwargs) |
| return future |
| |
| return rpc_call |
| |
| def __init__(self, |
| uid, |
| serial, |
| on_error_callback, |
| _create_connection_func, |
| max_connections=None): |
| """Creates a new RpcClient object. |
| |
| Args: |
| uid: The session uid this client is a part of. |
| serial: The serial of the Android device. Used for logging. |
| on_error_callback: A callback for when a connection error is raised. |
| _create_connection_func: A reference to the function that creates a |
| new session. |
| max_connections: The maximum number of connections the RpcClient |
| can have. |
| """ |
| self._serial = serial |
| self.on_error = on_error_callback |
| self._create_connection_func = _create_connection_func |
| self._free_connections = [self._create_connection_func(uid)] |
| |
| self.uid = self._free_connections[0].uid |
| self._lock = threading.Lock() |
| |
| def _log_formatter(message): |
| """Formats the message to be logged.""" |
| return '[RPC Service|%s|%s] %s' % (self._serial, self.uid, message) |
| |
| self._log = logger.create_logger(_log_formatter) |
| |
| self._working_connections = [] |
| if max_connections is None: |
| self.max_connections = RpcClient.DEFAULT_MAX_CONNECTION |
| else: |
| self.max_connections = max_connections |
| |
| self._async_client = RpcClient.AsyncClient(self) |
| self.is_alive = True |
| |
| def terminate(self): |
| """Terminates all connections to the SL4A server.""" |
| if len(self._working_connections) > 0: |
| self._log.warning( |
| '%s connections are still active, and waiting on ' |
| 'responses.Closing these connections now.' % len( |
| self._working_connections)) |
| connections = self._free_connections + self._working_connections |
| for connection in connections: |
| self._log.debug( |
| 'Closing connection over ports %s' % connection.ports) |
| connection.close() |
| self._free_connections = [] |
| self._working_connections = [] |
| self.is_alive = False |
| |
| def _get_free_connection(self): |
| """Returns a free connection to be used for an RPC call. |
| |
| This function also adds the client to the working set to prevent |
| multiple users from obtaining the same client. |
| """ |
| while True: |
| if len(self._free_connections) > 0: |
| with self._lock: |
| # Check if another thread grabbed the remaining connection. |
| # while we were waiting for the lock. |
| if len(self._free_connections) == 0: |
| continue |
| client = self._free_connections.pop() |
| self._working_connections.append(client) |
| return client |
| |
| client_count = ( |
| len(self._free_connections) + len(self._working_connections)) |
| if client_count < self.max_connections: |
| with self._lock: |
| client_count = (len(self._free_connections) + |
| len(self._working_connections)) |
| if client_count < self.max_connections: |
| client = self._create_connection_func(self.uid) |
| self._working_connections.append(client) |
| return client |
| time.sleep(.01) |
| |
| def _release_working_connection(self, connection): |
| """Marks a working client as free. |
| |
| Args: |
| connection: The client to mark as free. |
| Raises: |
| A ValueError if the client is not a known working connection. |
| """ |
| # We need to keep this code atomic because the client count is based on |
| # the length of the free and working connection list lengths. |
| with self._lock: |
| self._working_connections.remove(connection) |
| self._free_connections.append(connection) |
| |
| def rpc(self, method, *args, timeout=None, retries=1): |
| """Sends an rpc to sl4a. |
| |
| Sends an rpc call to sl4a over this RpcClient's corresponding session. |
| |
| Args: |
| method: str, The name of the method to execute. |
| args: any, The args to send to sl4a. |
| timeout: The amount of time to wait for a response. |
| retries: Misnomer, is actually the number of tries. |
| |
| Returns: |
| The result of the rpc. |
| |
| Raises: |
| Sl4aProtocolError: Something went wrong with the sl4a protocol. |
| Sl4aApiError: The rpc went through, however executed with errors. |
| """ |
| connection = self._get_free_connection() |
| ticket = connection.get_new_ticket() |
| |
| if timeout: |
| connection.set_timeout(timeout) |
| data = {'id': ticket, 'method': method, 'params': args} |
| request = json.dumps(data) |
| response = '' |
| try: |
| for i in range(1, retries + 1): |
| connection.send_request(request) |
| self._log.debug('Sent: %s' % request) |
| |
| response = connection.get_response() |
| self._log.debug('Received: %s', response) |
| if not response: |
| if i < retries: |
| self._log.warning( |
| 'No response for RPC method %s on iteration %s', |
| method, i) |
| continue |
| else: |
| self.on_error(connection) |
| raise Sl4aProtocolError( |
| Sl4aProtocolError.NO_RESPONSE_FROM_SERVER) |
| else: |
| break |
| except BrokenPipeError as e: |
| if self.is_alive: |
| self._log.error('Exception %s happened while communicating to ' |
| 'SL4A.', e) |
| self.on_error(connection) |
| else: |
| self._log.warning('The connection was killed during cleanup:') |
| self._log.warning(e) |
| raise Sl4aConnectionError(e) |
| finally: |
| if timeout: |
| connection.set_timeout(SOCKET_TIMEOUT) |
| self._release_working_connection(connection) |
| result = json.loads(str(response, encoding='utf8')) |
| |
| if result['error']: |
| err_msg = 'RPC call %s to device failed with error %s' % ( |
| method, result['error']) |
| self._log.error(err_msg) |
| raise Sl4aApiError(err_msg) |
| if result['id'] != ticket: |
| self._log.error('RPC method %s with mismatched api id %s', method, |
| result['id']) |
| raise Sl4aProtocolError(Sl4aProtocolError.MISMATCHED_API_ID) |
| return result['result'] |
| |
| @property |
| def future(self): |
| """Returns a magic function that returns a future running an RPC call. |
| |
| This function effectively allows the idiom: |
| |
| >>> rpc_client = RpcClient(...) |
| >>> # returns after call finishes |
| >>> rpc_client.someRpcCall() |
| >>> # Immediately returns a reference to the RPC's future, running |
| >>> # the lengthy RPC call on another thread. |
| >>> future = rpc_client.future.someLengthyRpcCall() |
| >>> rpc_client.doOtherThings() |
| >>> ... |
| >>> # Wait for and get the returned value of the lengthy RPC. |
| >>> # Can specify a timeout as well. |
| >>> value = future.result() |
| |
| The number of concurrent calls to this method is limited to |
| (max_connections - 2), to prevent future calls from exhausting all free |
| connections. |
| """ |
| return self._async_client |
| |
| def __getattr__(self, name): |
| """Wrapper for python magic to turn method calls into RPC calls.""" |
| |
| def rpc_call(*args, **kwargs): |
| return self.rpc(name, *args, **kwargs) |
| |
| if not self.is_alive: |
| raise Sl4aStartError( |
| 'This SL4A session has already been terminated. You must ' |
| 'create a new session to continue.') |
| return rpc_call |
