Streaming bidireccional en gRPC con Go

Curso de Go Avanzado: Protobuffers y gRPC

Contenido del curso

Introducción

gRPC

Conclusión

Tomar examen

Cursos Empresas Blog Live Conf Eventos Precios

0:0016:04

Streaming bidireccional en gRPC con Go

Curso de Go Avanzado: Protobuffers y gRPC

Clase 17 de 22

Clase 17 / 22

Recursos

El streaming bidireccional con gRPC te permite que cliente y servidor envíen y reciban datos al mismo tiempo a través de un único canal. Aquí aprendes a construir una función TakeTest en Go donde el servidor envía preguntas y el cliente responde, todo dentro del mismo flujo. Es ideal si trabajas en backend y quieres dominar comunicación en tiempo real.

¿Qué es el streaming bidireccional en gRPC?

Es una técnica que combina el streaming del lado del cliente y del lado del servidor en un solo canal full duplex. Eso significa que ambas partes pueden enviar y recibir mensajes simultáneamente sin abrir conexiones distintas.

¿Qué significa full duplex en gRPC? Es la capacidad de un mismo stream para enviar y recibir datos al mismo tiempo, sin necesidad de abrir dos canales separados.

En el caso práctico que vamos a construir, el servidor enviará las preguntas de un test y el cliente responderá cada una. Todo viaja por el mismo stream [00:14].

¿Cómo defino un RPC bidireccional en el archivo proto?

Lo primero es declarar el contrato en el archivo .proto dentro de la carpeta test.pb. Aquí marcas que tanto el parámetro de entrada como el de salida son streams.

proto rpc TakeTest(stream TakeTestRequest) returns (stream Question);

message TakeTestRequest { string answer = 1; }

La palabra clave stream aparece en ambos lados: en el request y en el response. Eso es lo que le indica a gRPC que la lectura y la escritura ocurrirán a través de data streaming [01:05]. El mensaje TakeTestRequest lleva un único campo answer, que es la respuesta que el cliente envía desde el front.

¿Cómo agrego la consulta de preguntas al repositorio?

Necesitas una función que devuelva las preguntas asociadas a un test específico. Esto se hace en dos capas: la interfaz del repositorio y la implementación concreta en PostgreSQL.

Interfaz abstracta del repositorio

En repository.go añades el método a la interfaz:

go GetQuestionsPerTest(ctx context.Context, testID string) ([]*models.Question, error)

Luego, en la capa abstracta, simplemente delegas la llamada a implementation.GetQuestionsPerTest. Es un patrón limpio que separa el contrato de la implementación [02:30].

Implementación concreta en PostgreSQL

Dentro de database/postgres.go creas la función sobre el tipo PostgresRepository. La consulta selecciona el id y la question desde la tabla questions filtrando por test_id.

go func (repo *PostgresRepository) GetQuestionsPerTest(ctx context.Context, testID string) ([]*models.Question, error) { rows, err := repo.db.QueryContext(ctx, "SELECT id, question FROM questions WHERE test_id = $1", testID) if err != nil { return nil, err } defer func() { if err := rows.Close(); err != nil { log.Fatal(err) } }()

var questions []*models.Question
for rows.Next() {
    var question = models.Question{}
    if err = rows.Scan(&question.Id, &question.Question); err == nil {
        questions = append(questions, &question)
    }
}
if err = rows.Err(); err != nil {
    return nil, err
}
return questions, nil

}

Usa defer con una función anónima para cerrar los rows y captura cualquier error de cierre con log.Fatal. El rows.Scan mapea las columnas a la estructura models.Question [03:50].

¿Cómo implemento TakeTest en el servidor gRPC?

En server/test.go defines la función TakeTest sobre el TestServer. Y aquí viene lo interesante: recibes un solo parámetro stream, no dos.

go func (s *TestServer) TakeTest(stream testpb.TestService_TakeTestServer) error { questions, err := s.repo.GetQuestionsPerTest(context.Background(), "t1") if err != nil { return err }

i := 0
var currentQuestion = &models.Question{}
for {
    if i < len(questions) {
        currentQuestion = questions[i]
    }
    if i <= len(questions) {
        questionToSend := &testpb.Question{
            Id:       currentQuestion.Id,
            Question: currentQuestion.Question,
        }
        if err := stream.Send(questionToSend); err != nil {
            return err
        }
        i++
    }

    answer, err := stream.Recv()
    if err == io.EOF {
        return nil
    }
    if err != nil {
        return err
    }
    log.Println("Answer received:", answer.Answer)
}

}

Antes de que el autocompletado funcione, debes recompilar los archivos .proto con el comando que generaste el paquete de Go. Tras compilar, el tipo TestService_TakeTestServer queda disponible con sus métodos Send y Recv, las dos piezas que necesita el streaming bidireccional [06:20].

¿Cuál es la diferencia entre Send y Recv en gRPC? Send enruta un mensaje hacia el otro extremo del stream; Recv lee un mensaje entrante. Ambos operan sobre el mismo objeto stream.

¿Por qué uso una variable contadora?

La variable i controla qué pregunta toca enviar y evita que mandes más preguntas de las que hay en la base de datos. La variable currentQuestion guarda la pregunta activa que se enviará al cliente en cada iteración del bucle. La validación i < len(questions) impide accesos fuera de rango.

¿Cómo manejo el fin del stream?

Cuando stream.Recv() devuelve io.EOF, significa que el cliente cerró su lado del canal. Ahí retornas nil para terminar TakeTest limpiamente. Cualquier otro error se propaga como retorno de la función [09:15].

Puntos clave del streaming bidireccional

Lo que distingue a esta técnica:

Un único objeto stream maneja envío y recepción.
El servidor empuja preguntas con stream.Send y lee respuestas con stream.Recv.
El cliente hace lo mismo en sentido contrario.
El cierre se detecta con io.EOF, no con timeouts.

Es full duplex puro: cada extremo decide cuándo hablar y cuándo escuchar. ¿Te animas a hacer configurable el testID en lugar de dejarlo fijo en "t1"? Cuéntame en los comentarios cómo lo resolverías.

Àlex Grau Roca

Hace 4 años

Les dejo en mi repositorio de GitHub el proyecto, con la opción de leer el test que se quiere realizar, saber cuando se ha finalizado el test y poder realizar otro test. . No tengo el código dividido exactamente como el profesor, ya que he estado creando ficheros específicos para cumplir un objetivo más concreto. Creo que así el código queda más limpio y fácil de gestionar, que no tener un único fichero con todas las funcionalidas. La excepción son los ficheros proto, ya que no sé si puede afectar o no (tendré que hacer alguna prueba/investigar). . Les resumo los cambios: .

testpb/test.proto: he reemplazado la respuesta de Question por un QuestionPerTest. Además, le añado el test_id al message TakeTest. El message QuestionPerTest es el siguiente:

message QuestionPerTest {
    string id = 1;
    string question = 2;
    bool ok = 3;
}

repository/TestServer.go: hago un mix entre lo que hace el profesor y el SetQuestions/EnrollStudents. Empiezo con un bucle for para leer el mensaje del cliente, en este mensaje se espera el id del test. A partir de aquí es prácticamente lo mismo que hace el profesor, con algunos ajustes. . Primero, en lugar de hacer siempre el len(questions), guardo el resultado en una variable para optimizar el código. En el questionToSend, le indico el campo OK a false. El segundo if lo elimino, realizando todo en un único if ya que no tiene sentido realizar dos veces la misma validación (además que el segundo if bajo mi punto de vista es incorrecto al utilizar el <=).
. A continuación añado un else para cuando ya se han mandado todas las questions. Cuando se procesa ese else lo que hago es devolver un questionToSend vacío, pero con el flag OK a true. Y muy importante, hay que añadirle un break al final, de este modo salimos del for infinito interno y dejamos la aplicación pendiente de recibir el siguiente test a realizar.

. Tras estos ajustes, el stream esperará a recibir primero el id del test para empezar a mandar questions y tras enviarlas todas, el cliente recibirá un ok = true, dejando el servidor en espera de recibir otro id de test a realizar. Algo que se me ha ocurrido redactando esto es que también se puede añadir el número de question actual y el total de questions en el QuestionPerTest, de este modo el cliente puede saber en todo momento la cantidad de questions que ha respondido y las que le faltan por responder.

Luis Vicente Arevalo Rios

Hace 4 años

se harán canciones y poesía en tu nombre, gracias Alex

Juan Diego Castellanos Jerez

Hace 3 años

Man! You're on Fire !!

Introducción

Microservicios de alto rendimiento con gRPC y Go

Proyecto gRPC con streaming bidireccional

Qué son los Protocol Buffers y cómo funcionan

Compilar Protobuf a paquete Go con protoc

De monolito a microservicios con gRPC

gRPC

gRPC: qué lo hace más rápido que REST

JSON vs Protobuf: cuándo usar cada uno

Definir un StudentService con proto buffers en Go

Implementación de Repositorios y gRPC en Go con PostgreSQL

Criando o primeiro servidor gRPC em Go

Prueba tu servidor gRPC con Postman

Definiendo el servicio de test en gRPC

Implementando servidor gRPC con PostgreSQL en Go

Client streaming en gRPC con Go

Implementación de Streaming del Servidor con gRPC y Protobufs

Streaming del servidor con gRPC en Go

Streaming bidireccional en gRPC con Go

Prueba de streaming bidireccional con gRPC

Implementando un cliente gRPC en Go

Streaming bidireccional en gRPC con Go

Conclusión

gRPC en el navegador con Envoy

Streaming de Datos con gRPC y Protobufers en Backends

Streaming bidireccional en gRPC con Go

Progreso del curso

Introducción

gRPC

Conclusión

Recursos

Recursos