gRPC e Protobuf in Flutter: client tipizzato, interceptor e streaming bidirezionale

gRPC in Dart richiede due cose distinte: le librerie runtime (dichiarate nel pubspec.yaml) e il compilatore protoc con il plugin Dart, che gira solo in fase di sviluppo.

1. Installa protoc (su macOS brew install protobuf, su Debian/Ubuntu apt install -y protobuf-compiler, su Windows scarica il binario dalle release di protocolbuffers).
2. Installa globalmente il plugin Dart e assicurati che ~/.pub-cache/bin sia nel PATH.
3. Aggiungi grpc e protobuf alle dipendenze dell'app.

Organizza il progetto separando il contratto dal codice generato: i file in lib/src/generated/ non vanno modificati a mano e conviene escluderli dall'analisi statica per non inquinare i report di lint.

# 1) Plugin protoc per Dart
dart pub global activate protoc_plugin
export PATH="$PATH:$HOME/.pub-cache/bin"

# 2) Dipendenze nell'app Flutter
flutter pub add grpc protobuf
flutter pub add --dev test

# 3) Struttura consigliata
# proto/chat.proto                 <- contratto (unica fonte di verità)
# lib/src/generated/               <- codice generato (non versionare le modifiche manuali)
# lib/src/data/grpc/               <- channel, interceptor, repository

# 4) analysis_options.yaml
# analyzer:
#   exclude:
#     - lib/src/generated/**

Il file .proto è il confine del modulo: descrive messaggi e servizi senza alcun riferimento a Flutter. Usiamo proto3, includiamo una chiamata unaria, una server-streaming e una bidirezionale.

Alcune regole pratiche per evitare breaking change:

• i numeri di campo sono il vero contratto: non riusarli mai dopo una rimozione (usa reserved);
• preferisci google.protobuf.Timestamp ai timestamp in stringa;
• per i campi opzionali in proto3 usa optional (genera hasX()), altrimenti non distingui "assente" da "valore di default".

La generazione va automatizzata in uno script, così è riproducibile anche in CI: ogni modifica al .proto deve passare da tool/protoc.sh, mai da editing manuale dei file generati.

// proto/chat.proto
syntax = "proto3";
package chat.v1;

import "google/protobuf/timestamp.proto";

message User {
  string id = 1;
  string display_name = 2;
  optional string avatar_url = 3;
}

message GetUserRequest { string id = 1; }

message Message {
  string id = 1;
  string room_id = 2;
  User author = 3;
  string text = 4;
  google.protobuf.Timestamp sent_at = 5;
}

message WatchRequest { string room_id = 1; }
message SendRequest { string room_id = 1; string text = 2; }
message SendAck { string message_id = 1; }

service ChatService {
  rpc GetUser(GetUserRequest) returns (User);
  rpc WatchMessages(WatchRequest) returns (stream Message);
  rpc Chat(stream SendRequest) returns (stream Message);
}

// --- tool/protoc.sh ---
// #!/usr/bin/env bash
// set -euo pipefail
// OUT=lib/src/generated
// rm -rf "$OUT" && mkdir -p "$OUT"
// protoc --dart_out=grpc:$OUT -Iproto proto/*.proto
// dart format "$OUT"

Il ClientChannel è una risorsa costosa e condivisa: va creato una sola volta per host e chiuso allo shutdown dell'app. Evita di istanziarlo dentro un widget o a ogni chiamata.

Punti chiave della configurazione:

• Credenziali: ChannelCredentials.secure() in produzione, insecure() solo verso localhost in sviluppo (ricorda che su Android l'emulatore vede l'host come 10.0.2.2).
• Keepalive: indispensabile su mobile, dove NAT e radio chiudono connessioni idle; invia PING periodici per tenere viva la connessione HTTP/2.
• Deadline: gRPC non ha un "timeout globale", si usa CallOptions(timeout: ...) per chiamata. Le unarie devono sempre avere una deadline; gli stream long-lived no, altrimenti vengono chiusi.
• Idle timeout: lascia che il canale si spenga quando inattivo per risparmiare batteria.

import 'package:grpc/grpc.dart';
import '../generated/chat.pbgrpc.dart';

class GrpcEnvironment {
  const GrpcEnvironment({
    required this.host,
    required this.port,
    required this.useTls,
  });

  final String host;
  final int port;
  final bool useTls;
}

class GrpcClientFactory {
  GrpcClientFactory(this._env, this._interceptors);

  final GrpcEnvironment _env;
  final List<ClientInterceptor> _interceptors;
  ClientChannel? _channel;

  ClientChannel get channel => _channel ??= ClientChannel(
        _env.host,
        port: _env.port,
        options: ChannelOptions(
          credentials: _env.useTls
              ? const ChannelCredentials.secure()
              : const ChannelCredentials.insecure(),
          idleTimeout: const Duration(minutes: 5),
          connectionTimeout: const Duration(seconds: 10),
          keepAlive: const ClientKeepAliveOptions(
            pingInterval: Duration(seconds: 30),
            timeout: Duration(seconds: 10),
            permitWithoutCalls: false,
          ),
          userAgent: 'flutter-chat/1.0',
        ),
      );

  ChatServiceClient get chat => ChatServiceClient(
        channel,
        interceptors: _interceptors,
        options: CallOptions(timeout: const Duration(seconds: 15)),
      );

  Future<void> dispose() async {
    await _channel?.shutdown();
    _channel = null;
  }
}

Un ClientInterceptor intercetta tutte le chiamate (unarie e streaming) prima che partano: è il posto giusto per header di autenticazione, correlation id e logging dei tempi.

Due strategie per i metadata:

1. CallOptions(providers: [...]): callback asincrona eseguita a ogni chiamata, ideale perché legge il token aggiornato dal secure storage.
2. Merge di CallOptions dentro l'interceptor, utile per header statici (versione app, locale).

Per il refresh del token non fare retry dentro l'interceptor della stessa chiamata: intercetta StatusCode.unauthenticated nel layer repository, rinnova il token e ripeti l'invocazione una sola volta, proteggendo il refresh con un Completer condiviso per evitare chiamate concorrenti.

import 'dart:developer' as dev;
import 'package:grpc/grpc.dart';

typedef TokenProvider = Future<String?> Function();

class AuthInterceptor extends ClientInterceptor {
  AuthInterceptor(this._tokenProvider, {this.appVersion = '1.0.0'});

  final TokenProvider _tokenProvider;
  final String appVersion;

  CallOptions _withAuth(CallOptions options) => options.mergedWith(
        CallOptions(
          metadata: {'x-app-version': appVersion},
          providers: [
            (Map<String, String> metadata, String uri) async {
              final token = await _tokenProvider();
              if (token != null && token.isNotEmpty) {
                metadata['authorization'] = 'Bearer $token';
              }
            },
          ],
        ),
      );

  @override
  ResponseFuture<R> interceptUnary<Q, R>(
    ClientMethod<Q, R> method,
    Q request,
    CallOptions options,
    ClientUnaryInvoker<Q, R> invoker,
  ) =>
      invoker(method, request, _withAuth(options));

  @override
  ResponseStream<R> interceptStreaming<Q, R>(
    ClientMethod<Q, R> method,
    Stream<Q> requests,
    CallOptions options,
    ClientStreamingInvoker<Q, R> invoker,
  ) =>
      invoker(method, requests, _withAuth(options));
}

class LoggingInterceptor extends ClientInterceptor {
  @override
  ResponseFuture<R> interceptUnary<Q, R>(
    ClientMethod<Q, R> method,
    Q request,
    CallOptions options,
    ClientUnaryInvoker<Q, R> invoker,
  ) {
    final sw = Stopwatch()..start();
    final call = invoker(method, request, options);
    call.then(
      (_) => dev.log('${method.path} ok in ${sw.elapsedMilliseconds}ms', name: 'grpc'),
      onError: (Object e) =>
          dev.log('${method.path} KO in ${sw.elapsedMilliseconds}ms: $e', name: 'grpc'),
    );
    return call;
  }
}

GrpcError espone un code numerico standard: trasformarlo in un errore di dominio evita che package:grpc si propaghi fino alla UI.

Regola d'oro del retry: ritenta solo ciò che è sicuro ritentare. unavailable e deadlineExceeded sono tipicamente transitori; invalidArgument, permissionDenied, notFound no. Per le chiamate non idempotenti (es. invio messaggio) usa una idempotency key nei metadata, così il server può deduplicare i tentativi.

Il backoff deve essere esponenziale con jitter: senza jitter, tutti i client riconnettono nello stesso istante dopo un disservizio, generando un thundering herd.

import 'dart:async';
import 'dart:math';
import 'package:grpc/grpc.dart';

sealed class AppFailure implements Exception {
  const AppFailure(this.message);
  final String message;
}

class NetworkFailure extends AppFailure { const NetworkFailure(super.m); }
class AuthFailure extends AppFailure { const AuthFailure(super.m); }
class NotFoundFailure extends AppFailure { const NotFoundFailure(super.m); }
class ServerFailure extends AppFailure { const ServerFailure(super.m); }

AppFailure mapGrpcError(GrpcError e) => switch (e.code) {
      StatusCode.unavailable || StatusCode.deadlineExceeded =>
        const NetworkFailure('Server non raggiungibile, riprova.'),
      StatusCode.unauthenticated || StatusCode.permissionDenied =>
        const AuthFailure('Sessione non valida.'),
      StatusCode.notFound => const NotFoundFailure('Risorsa inesistente.'),
      _ => ServerFailure(e.message ?? 'Errore imprevisto (${e.code}).'),
    };

const _retryable = {StatusCode.unavailable, StatusCode.deadlineExceeded, StatusCode.resourceExhausted};
final _rnd = Random();

Future<T> retryUnary<T>(
  Future<T> Function() action, {
  int maxAttempts = 3,
  Duration base = const Duration(milliseconds: 300),
}) async {
  var attempt = 0;
  while (true) {
    attempt++;
    try {
      return await action();
    } on GrpcError catch (e) {
      final canRetry = _retryable.contains(e.code) && attempt < maxAttempts;
      if (!canRetry) throw mapGrpcError(e);
      final expo = base * pow(2, attempt - 1).toInt();
      final jitter = Duration(milliseconds: _rnd.nextInt(250));
      await Future<void>.delayed(expo + jitter);
    }
  }
}

class ChatRepository {
  ChatRepository(this._factory);
  final GrpcClientFactory _factory;

  Future<User> getUser(String id) => retryUnary(
        () => _factory.chat
            .getUser(GetUserRequest(id: id),
                options: CallOptions(timeout: const Duration(seconds: 8)))
            .then((u) => u),
      );
}

Le rpc ... returns (stream ...) producono una ResponseStream: un Stream Dart arricchito con headers, trailers e cancel().

Per un flusso realtime affidabile su mobile servono tre accorgimenti:

1. Nessuna deadline sullo stream long-lived (timeout: null), altrimenti verrà chiuso dal client.
2. Riconnessione con backoff quando il server chiude con unavailable: incapsula lo stream in uno StreamController che ricrea la sottoscrizione, passando l'ultimo cursore ricevuto (lastMessageId) per non perdere eventi.
3. Cancellazione deterministica: quando l'ascoltatore si disiscrive (onCancel) devi chiamare cancel() sulla ResponseStream, altrimenti la chiamata resta aperta lato server.

Usa broadcast: false e lascia che sia la UI a sottoscriversi tramite StreamBuilder: il ciclo di vita del widget guida quello della RPC.

import 'dart:async';
import 'package:grpc/grpc.dart';

extension ChatStreams on ChatRepository {
  Stream<Message> watchMessages(String roomId) {
    late StreamController<Message> controller;
    StreamSubscription<Message>? sub;
    ResponseStream<Message>? call;
    var attempt = 0;
    var closed = false;

    Future<void> connect() async {
      if (closed) return;
      call = _factory.chat.watchMessages(
        WatchRequest(roomId: roomId),
        options: CallOptions(timeout: null), // stream long-lived
      );
      sub = call!.listen(
        (msg) {
          attempt = 0; // connessione sana: azzera il backoff
          controller.add(msg);
        },
        onError: (Object e, StackTrace st) async {
          if (e is GrpcError && e.code == StatusCode.unavailable) {
            attempt++;
            if (attempt > 5) {
              controller.addError(mapGrpcError(e), st);
              await controller.close();
              return;
            }
            await Future<void>.delayed(Duration(seconds: 1 << (attempt - 1)));
            await connect();
          } else {
            controller.addError(
              e is GrpcError ? mapGrpcError(e) : e,
              st,
            );
          }
        },
        onDone: () async {
          if (!closed) await connect();
        },
        cancelOnError: false,
      );
    }

    controller = StreamController<Message>(
      onListen: connect,
      onCancel: () async {
        closed = true;
        await sub?.cancel();
        await call?.cancel();
      },
    );
    return controller.stream;
  }
}

// Uso nella UI
// StreamBuilder<Message>(
//   stream: repository.watchMessages(roomId),
//   builder: (context, snapshot) {
//     if (snapshot.hasError) return ErrorView(snapshot.error! as AppFailure);
//     ...
//   },
// )

Il vantaggio di gRPC in test è che puoi avviare un vero server in-process: niente mock fragili, verifichi anche serializzazione, metadata e codici di stato.

Nel test:

• implementa ChatServiceBase con comportamenti controllati (errori, stream finiti, ritardi);
• avvia Server.create(...) su port: 0 e leggi server.port;
• punta il ClientChannel a 127.0.0.1 con credenziali insecure.

Per la CI ricorda di eseguire tool/protoc.sh e fallire la build se git diff --exit-code lib/src/generated rileva codice generato non allineato al .proto.

Flutter Web: il browser non espone i frame HTTP/2 necessari a gRPC. Serve un proxy che parli gRPC-Web (Envoy o grpcwebproxy) e, lato Dart, GrpcWebClientChannel.xhr(Uri.parse('https://api.example.com')) al posto di ClientChannel. Astrai la creazione del canale dietro un'interfaccia con implementazioni condizionali (io.dart / web.dart) così il repository resta identico su tutte le piattaforme. Nota che lo streaming bidirezionale non è supportato da gRPC-Web: su web ripiega su server-streaming o WebSocket.

import 'package:grpc/grpc.dart';
import 'package:test/test.dart';

class FakeChatService extends ChatServiceBase {
  int getUserCalls = 0;

  @override
  Future<User> getUser(ServiceCall call, GetUserRequest request) async {
    getUserCalls++;
    final auth = call.clientMetadata?['authorization'];
    if (auth != 'Bearer test-token') {
      throw GrpcError.unauthenticated('token mancante');
    }
    if (getUserCalls < 2) {
      throw GrpcError.unavailable('server in riavvio'); // forza un retry
    }
    return User(id: request.id, displayName: 'Ada');
  }

  @override
  Stream<Message> watchMessages(ServiceCall call, WatchRequest request) async* {
    yield Message(id: '1', roomId: request.roomId, text: 'ciao');
    yield Message(id: '2', roomId: request.roomId, text: 'mondo');
  }

  @override
  Stream<Message> chat(ServiceCall call, Stream<SendRequest> requests) async* {
    await for (final r in requests) {
      yield Message(id: 'echo', roomId: r.roomId, text: r.text);
    }
  }
}

void main() {
  late Server server;
  late GrpcClientFactory factory;

  setUp(() async {
    server = Server.create(services: [FakeChatService()]);
    await server.serve(address: '127.0.0.1', port: 0);
    factory = GrpcClientFactory(
      GrpcEnvironment(host: '127.0.0.1', port: server.port!, useTls: false),
      [AuthInterceptor(() async => 'test-token')],
    );
  });

  tearDown(() async {
    await factory.dispose();
    await server.shutdown();
  });

  test('retry su unavailable e mapping del risultato', () async {
    final user = await ChatRepository(factory).getUser('u1');
    expect(user.displayName, 'Ada');
  });

  test('server streaming emette tutti i messaggi', () async {
    final messages = await ChatRepository(factory).watchMessages('room-1').take(2).toList();
    expect(messages.map((m) => m.text), ['ciao', 'mondo']);
  });
}