04 - Le streaming : la vraie force de gRPC

Ce que tu vas apprendre

• Les 4 patterns de communication gRPC
• Implementer chaque type de streaming en TypeScript
• Les cas d'usage réels pour chaque pattern
• Pourquoi REST ne peut pas rivaliser sur ce terrain

Prerequisites

03 - Le client gRPC en TypeScript

Pourquoi le streaming change la donne

En REST, la communication est simple : une requête, une réponse. Si tu veux des donnees en temps réel, tu ajoutes du WebSocket ou du Server-Sent Events (SSE) a cote. Deux protocoles, deux formats, deux systèmes d'erreurs a maintenir.

En gRPC, le streaming fait partie du framework. Meme protocole, meme format, meme gestion d'erreurs. Et ca marche dans les deux sens.

Le fichier .proto

protobuf// proto/monitoring.proto
syntax = "proto3";

package monitoring;

service MonitoringService {
  // 1. Unary : une requete, une reponse
  rpc GetMetric (GetMetricRequest) returns (Metric);

  // 2. Server streaming : une requete, un flux de reponses
  rpc StreamMetrics (StreamMetricsRequest) returns (stream Metric);

  // 3. Client streaming : un flux de requetes, une reponse
  rpc UploadMetrics (stream Metric) returns (UploadMetricsResponse);

  // 4. Bidirectional streaming : flux dans les deux sens
  rpc LiveDashboard (stream DashboardFilter) returns (stream Metric);
}

message Metric {
  string name = 1;
  double value = 2;
  string unit = 3;
  int64 timestamp = 4;
  map<string, string> labels = 5;
}

message GetMetricRequest {
  string name = 1;
}

message StreamMetricsRequest {
  string name = 1;
  int32 interval_seconds = 2;
}

message UploadMetricsResponse {
  int32 received_count = 1;
  bool success = 2;
}

message DashboardFilter {
  repeated string metric_names = 1;
  double threshold = 2;
}

Le mot-clé stream devant un type indique que c'est un flux, pas un message unique. C'est la seule différence syntaxique avec un RPC unaire.

Pattern 1 : Unary (requête/réponse)

C'est le mode classique, identique a un appel REST. On l'a deja couvert dans les articles précédents. Une requête, une réponse, c'est fini.

typescript// Serveur
getMetric(call, callback) {
  const metric = getCurrentMetric(call.request.name);
  callback(null, metric);
}

// Client
const metric = await grpcPromise(client.getMetric.bind(client), { name: "cpu_usage" });

Cas d'usage : lectures simples, CRUD, tout ce qui n'a pas besoin de flux continu.

Pattern 2 : Server streaming

Le client envoie une requête. Le serveur répond avec un flux de messages qui arrive au fur et a mesure. Le client lit les messages un par un.

typescript// Serveur : envoie des metriques toutes les N secondes
streamMetrics(call) {
  const { name, intervalSeconds } = call.request;
  const interval = (intervalSeconds || 5) * 1000;

  const timer = setInterval(() => {
    const metric = {
      name,
      value: Math.random() * 100,
      unit: "percent",
      timestamp: Date.now(),
      labels: { host: "server-01" }
    };
    call.write(metric);
  }, interval);

  // Quand le client se deconnecte
  call.on("cancelled", () => {
    clearInterval(timer);
  });

  // Pour arreter le stream cote serveur
  // call.end();
}

typescript// Client : lit le flux
const stream = client.streamMetrics({ name: "cpu_usage", intervalSeconds: 2 });

stream.on("data", (metric) => {
  console.log(`${metric.name}: ${metric.value.toFixed(1)}${metric.unit} @ ${metric.timestamp}`);
});

stream.on("end", () => {
  console.log("Stream termine");
});

stream.on("error", (err) => {
  console.error("Erreur stream:", err.message);
});

Cas d'usage réels :

• Suivre les logs d'un service en temps réel
• Dashboard de metriques qui se met à jour en continu
• Notifications push du serveur vers un service
• Telecharger un gros fichier par morceaux

C'est l'équivalent de Server-Sent Events (SSE) en REST, mais avec le typage Protobuf et sans HTTP supplementaire.

Pattern 3 : Client streaming

Le client envoie un flux de messages. Le serveur attend de tout recevoir (ou traite au fur et a mesure) puis répond une seule fois.

typescript// Serveur : recoit un flux de metriques et repond un resume
uploadMetrics(call, callback) {
  let count = 0;

  call.on("data", (metric) => {
    count++;
    storeMetric(metric); // Persiste chaque metrique
  });

  call.on("end", () => {
    callback(null, {
      receivedCount: count,
      success: true
    });
  });

  call.on("error", (err) => {
    console.error("Erreur client stream:", err.message);
    callback({
      code: grpc.status.INTERNAL,
      message: "failed to process metrics"
    });
  });
}

typescript// Client : envoie un lot de metriques
const stream = client.uploadMetrics((err, response) => {
  if (err) {
    console.error("Upload echoue:", err.message);
    return;
  }
  console.log(`${response.receivedCount} metriques envoyees`);
});

// Envoie les metriques une par une
const metrics = collectLocalMetrics();
for (const metric of metrics) {
  stream.write(metric);
}

// Signale la fin de l'envoi
stream.end();

Cas d'usage réels :

• Upload de fichiers par morceaux (chunks)
• Envoi de metriques en batch depuis un agent
• Ingestion de logs depuis un collecteur
• Import massif de donnees

Pattern 4 : Bidirectional streaming

Les deux cotes envoient et recoivent en continu. C'est le mode le plus puissant et le plus complexe.

typescript// Serveur : le client envoie des filtres, le serveur renvoie les metriques matchantes
liveDashboard(call) {
  let currentFilters: string[] = [];
  let threshold = 0;

  // Le client envoie des filtres qui changent dynamiquement
  call.on("data", (filter) => {
    currentFilters = filter.metricNames;
    threshold = filter.threshold;
    console.log(`Filtres mis a jour: ${currentFilters.join(", ")} (seuil: ${threshold})`);
  });

  // Envoie des metriques qui correspondent aux filtres actuels
  const timer = setInterval(() => {
    for (const name of currentFilters) {
      const value = getMetricValue(name);
      if (value > threshold) {
        call.write({
          name,
          value,
          unit: "percent",
          timestamp: Date.now(),
          labels: { alert: "above_threshold" }
        });
      }
    }
  }, 1000);

  call.on("end", () => {
    clearInterval(timer);
    call.end();
  });

  call.on("cancelled", () => {
    clearInterval(timer);
  });
}

typescript// Client : envoie des filtres et recoit les metriques en temps reel
const stream = client.liveDashboard();

// Recevoir les metriques
stream.on("data", (metric) => {
  console.log(`ALERTE: ${metric.name} = ${metric.value} (seuil depasse)`);
});

stream.on("end", () => {
  console.log("Dashboard ferme");
});

// Envoyer les filtres initiaux
stream.write({
  metricNames: ["cpu_usage", "memory_usage"],
  threshold: 80
});

// 10 secondes plus tard, ajouter un filtre
setTimeout(() => {
  stream.write({
    metricNames: ["cpu_usage", "memory_usage", "disk_io"],
    threshold: 90
  });
}, 10000);

// Fermer apres 60 secondes
setTimeout(() => {
  stream.end();
}, 60000);

Cas d'usage réels :

• Chat en temps réel (chaque participant envoie et recoit des messages)
• Jeux multijoueur (positions, actions)
• Dashboard interactif (filtres dynamiques + metriques live)
• Synchronisation bidirectionnelle entre services

Streaming vs WebSocket vs SSE

gRPC streaming est supérieur pour la communication inter-services. WebSocket et SSE restent meilleurs pour la communication navigateur-serveur. Sur paltemps.fr, si on devait ajouter du temps réel (notifications de nouveaux articles), SSE serait le bon choix cote client. gRPC streaming n'a de sens que si on avait des services internes a connecter.

Le piège du streaming mal gere

Le streaming gRPC nécessité de la rigueur sur la gestion du cycle de vie :

typescript// TOUJOURS nettoyer les ressources
call.on("cancelled", () => {
  clearInterval(timer);
  closeDbConnection();
});

call.on("error", (err) => {
  clearInterval(timer);
  console.error("Stream error:", err);
});

// TOUJOURS gerer le backpressure
const canWrite = call.write(data);
if (!canWrite) {
  // Le buffer est plein, attends avant d'envoyer plus
  await new Promise(resolve => call.once("drain", resolve));
}

Si tu ne nettoies pas les timers et connexions quand le client se deconnecte, tu as des fuites de mémoire. Si tu ne geres pas le backpressure, tu satures le buffer et tu perds des messages.

Résumé

• Unary : requête/réponse classique, équivalent REST
• Server streaming : le serveur envoie un flux (logs, metriques, notifications)
• Client streaming : le client envoie un flux (upload, ingestion de donnees)
• Bidirectional : flux dans les deux sens (chat, dashboard interactif)
• Toujours nettoyer les ressources sur cancel/error
• gRPC streaming pour l'inter-service, WebSocket/SSE pour le navigateur

Article précédent : 03 - Le client gRPC Article suivant : 05 - gRPC en production

Sources

• gRPC - Core Concepts: RPC Life Cycle
• gRPC Node.js - Streaming
• HTTP/2 Flow Control