async def incoming_api_task(self) -> None:
    self.tasks = set()
    while True:
        (payload_inc, connection_inc) = (await self.incoming_messages.get())
        if ((payload_inc is None) or (connection_inc is None)):
            continue

        async def api_call(full_message: Message, connection: WSGreenBerryConnection, task_id):
            start_time = time.time()
            try:
                if (self.received_message_callback is not None):
                    (await self.received_message_callback(connection))
                connection.log.debug(f'<- {ProtocolMessageTypes(full_message.type).name} from peer {connection.peer_node_id} {connection.peer_host}')
                message_type: str = ProtocolMessageTypes(full_message.type).name
                f = getattr(self.api, message_type, None)
                if (f is None):
                    self.log.error(f'Non existing function: {message_type}')
                    raise ProtocolError(Err.INVALID_PROTOCOL_MESSAGE, [message_type])
                if (not hasattr(f, 'api_function')):
                    self.log.error(f'Peer trying to call non api function {message_type}')
                    raise ProtocolError(Err.INVALID_PROTOCOL_MESSAGE, [message_type])
                if hasattr(self.api, 'api_ready'):
                    if (self.api.api_ready is False):
                        return None
                timeout: Optional[int] = 600
                if hasattr(f, 'execute_task'):
                    self.execute_tasks.add(task_id)
                    timeout = None
                if hasattr(f, 'peer_required'):
                    coroutine = f(full_message.data, connection)
                else:
                    coroutine = f(full_message.data)

                async def wrapped_coroutine() -> Optional[Message]:
                    try:
                        result = (await coroutine)
                        return result
                    except asyncio.CancelledError:
                        pass
                    except Exception as e:
                        tb = traceback.format_exc()
                        connection.log.error(f'Exception: {e}, {connection.get_peer_info()}. {tb}')
                        raise e
                    return None
                response: Optional[Message] = (await asyncio.wait_for(wrapped_coroutine(), timeout=timeout))
                connection.log.debug(f'Time taken to process {message_type} from {connection.peer_node_id} is {(time.time() - start_time)} seconds')
                if (response is not None):
                    response_message = Message(response.type, full_message.id, response.data)
                    (await connection.reply_to_request(response_message))
            except Exception as e:
                if (self.connection_close_task is None):
                    tb = traceback.format_exc()
                    connection.log.error(f'Exception: {e} {type(e)}, closing connection {connection.get_peer_info()}. {tb}')
                else:
                    connection.log.debug(f'Exception: {e} while closing connection')
                (await connection.close(self.api_exception_ban_seconds, WSCloseCode.PROTOCOL_ERROR, Err.UNKNOWN))
            finally:
                if (task_id in self.api_tasks):
                    self.api_tasks.pop(task_id)
                if (task_id in self.tasks_from_peer[connection.peer_node_id]):
                    self.tasks_from_peer[connection.peer_node_id].remove(task_id)
                if (task_id in self.execute_tasks):
                    self.execute_tasks.remove(task_id)
        task_id = token_bytes()
        api_task = asyncio.create_task(api_call(payload_inc, connection_inc, task_id))
        self.api_tasks[task_id] = api_task
        if (connection_inc.peer_node_id not in self.tasks_from_peer):
            self.tasks_from_peer[connection_inc.peer_node_id] = set()
        self.tasks_from_peer[connection_inc.peer_node_id].add(task_id)
async def incoming_api_task(self) -> None:
    self.tasks = set()
    while True:
        (payload_inc, connection_inc) = (await self.incoming_messages.get())
        if ((payload_inc is None) or (connection_inc is None)):
            continue

        async def api_call(full_message: Message, connection: WSGreenBerryConnection, task_id):
            start_time = time.time()
            try:
                if (self.received_message_callback is not None):
                    (await self.received_message_callback(connection))
                connection.log.debug(f'<- {ProtocolMessageTypes(full_message.type).name} from peer {connection.peer_node_id} {connection.peer_host}')
                message_type: str = ProtocolMessageTypes(full_message.type).name
                f = getattr(self.api, message_type, None)
                if (f is None):
                    self.log.error(f'Non existing function: {message_type}')
                    raise ProtocolError(Err.INVALID_PROTOCOL_MESSAGE, [message_type])
                if (not hasattr(f, 'api_function')):
                    self.log.error(f'Peer trying to call non api function {message_type}')
                    raise ProtocolError(Err.INVALID_PROTOCOL_MESSAGE, [message_type])
                if hasattr(self.api, 'api_ready'):
                    if (self.api.api_ready is False):
                        return None
                timeout: Optional[int] = 600
                if hasattr(f, 'execute_task'):
                    self.execute_tasks.add(task_id)
                    timeout = None
                if hasattr(f, 'peer_required'):
                    coroutine = f(full_message.data, connection)
                else:
                    coroutine = f(full_message.data)

                async def wrapped_coroutine() -> Optional[Message]:
                    try:
                        result = (await coroutine)
                        return result
                    except asyncio.CancelledError:
                        pass
                    except Exception as e:
                        tb = traceback.format_exc()
                        connection.log.error(f'Exception: {e}, {connection.get_peer_info()}. {tb}')
                        raise e
                    return None
                response: Optional[Message] = (await asyncio.wait_for(wrapped_coroutine(), timeout=timeout))
                connection.log.debug(f'Time taken to process {message_type} from {connection.peer_node_id} is {(time.time() - start_time)} seconds')
                if (response is not None):
                    response_message = Message(response.type, full_message.id, response.data)
                    (await connection.reply_to_request(response_message))
            except Exception as e:
                if (self.connection_close_task is None):
                    tb = traceback.format_exc()
                    connection.log.error(f'Exception: {e} {type(e)}, closing connection {connection.get_peer_info()}. {tb}')
                else:
                    connection.log.debug(f'Exception: {e} while closing connection')
                (await connection.close(self.api_exception_ban_seconds, WSCloseCode.PROTOCOL_ERROR, Err.UNKNOWN))
            finally:
                if (task_id in self.api_tasks):
                    self.api_tasks.pop(task_id)
                if (task_id in self.tasks_from_peer[connection.peer_node_id]):
                    self.tasks_from_peer[connection.peer_node_id].remove(task_id)
                if (task_id in self.execute_tasks):
                    self.execute_tasks.remove(task_id)
        task_id = token_bytes()
        api_task = asyncio.create_task(api_call(payload_inc, connection_inc, task_id))
        self.api_tasks[task_id] = api_task
        if (connection_inc.peer_node_id not in self.tasks_from_peer):
            self.tasks_from_peer[connection_inc.peer_node_id] = set()
        self.tasks_from_peer[connection_inc.peer_node_id].add(task_id)