Tengo un protocolo de aplicación multiplexado. He implementado etapas de canalización para el procesamiento en clases de datos de Kotlin que representan cada tipo de mensaje en el protocolo, en ambas direcciones. No existe un patrón formal de solicitud-respuesta; el cliente y el servidor pueden enviarse mensajes en cualquier momento de la vida útil del socket.
En el protocolo, existe la restricción de que para cualquier cliente conectado, los mensajes se manejen en serie. Es decir, cuando el servidor recibe un mensaje de un cliente, no puede manejar el siguiente mensaje del cliente hasta que el anterior se haya manejado por completo. La misma regla se aplica al cliente que maneja los mensajes del servidor. Así es más o menos cómo se manejan las canalizaciones dentro de Netty; una etapa de canalización determinada solo puede manejar una cosa a la vez. Esto es bastante fácil de codificar al tener una etapa de canalización de bloqueo en un nuevo grupo de ejecutores.
Sin embargo, quiero escribir los controladores para cada tipo de mensaje que el cliente puede enviar en forma de suspensión de funciones, ya que el rango de cosas que mi servidor puede hacer en respuesta a un mensaje del cliente es muy amplio y puede implicar contactar a otros servidores o bases de datos. , y un grupo de tipos de mensajes también puede bloquear partes del estado del servidor internamente durante el procesamiento para excluir a otros clientes conectados simultáneamente. Por lo tanto, necesito una forma de vincularme a un contexto de rutina desde el final de la canalización del canal, de modo que la etapa de la canalización esté "bloqueada" para que no continúe procesando mensajes, pero no impida el procesamiento de las canalizaciones de otros canales en esta etapa, y sin crear un único ejecutor de eventos de hilo por socket.
Pensé en usar un despachador y un bucle de eventos separados de Netty para representar el procesamiento de cada sesión de cliente, usando un actor para representar cada uno, de modo que los mensajes recibidos se envíen al canal del buzón del actor y una corrutina los procese en el orden recibido, y Puedo cancelar el actor y en su cancelación cerrar el SocketChannel y viceversa. Pero esto es bastante feo y no se siente en el espíritu de Netty.
tl; dr Me pregunto, ¿cuál es la mejor manera de implementar una etapa de canalización que integre una función de suspensión limpiamente con su grupo ejecutor?