rpc_dart 7.0.0

Transport-agnostic RPC framework with contracts, streaming and included transports.

RPC Dart

Transport-agnostic RPC framework for Dart. One contract runs over any transport without code changes: in-process, isolates, WebSocket, HTTP/1.1, and HTTP/2 with the real gRPC wire protocol. This package is the core; transports other than the in-process ones live in separate packages (see Ecosystem).

---

Core concepts

• Contract — a service name plus its methods. A responder contract (RpcResponderContract) handles calls; a caller contract (RpcCallerContract) makes them.
• Endpoint — binds contracts to one transport: RpcResponderEndpoint (server), RpcCallerEndpoint (client), RpcPeerEndpoint (both directions).
• Transport — anything implementing IRpcTransport.
• Codec — serializes requests and responses. Optional for in-process zero-copy transports, required on the network.

Key features

• Unary, server streaming, client streaming, bidirectional streaming.
• Zero-copy in-process transport (RpcChannelTransport.memoryPair()).
• 3-layer transport architecture: IRpcChannel → IRpcMultiplexedChannel → IRpcTransport.
• Resilience: retry, circuit breaker, rate limiter, reconnecting client connection.
• Per-stream and connection-wide flow control, on by default.
• Security policy: bounds on metadata, message size and concurrent streams.
• gRPC Health Checking Protocol (grpc.health.v1).
• RpcBinaryCodec for protobuf and other binary formats.
• RpcContext with trace id, headers, deadline, cancellation.
• Pure Dart core — no runtime dependencies.

---

Quick start

Define the messages. RpcCodec serializes any IRpcSerializable (CBOR of toJson()):

import 'package:rpc_dart/rpc_dart.dart';

const calculatorService = 'Calculator';

class SumRequest implements IRpcSerializable {
  final List<double> values;
  SumRequest(this.values);

  factory SumRequest.fromJson(Map<String, dynamic> json) => SumRequest(
    (json['values'] as List).map((v) => (v as num).toDouble()).toList(),
  );

  @override
  Map<String, dynamic> toJson() => {'values': values};
}

class SumResponse implements IRpcSerializable {
  final double result;
  SumResponse(this.result);

  factory SumResponse.fromJson(Map<String, dynamic> json) =>
      SumResponse((json['result'] as num).toDouble());

  @override
  Map<String, dynamic> toJson() => {'result': result};
}

Implement the responder (server side) and the caller (client side):

class CalculatorResponder extends RpcResponderContract {
  CalculatorResponder() : super(calculatorService);

  @override
  void setup() {
    addUnaryMethod<SumRequest, SumResponse>(
      methodName: 'sum',
      requestCodec: RpcCodec(SumRequest.fromJson),
      responseCodec: RpcCodec(SumResponse.fromJson),
      handler: (request, {context}) async {
        final total = request.values.fold<double>(0, (a, b) => a + b);
        return SumResponse(total);
      },
    );
  }
}

class CalculatorCaller extends RpcCallerContract {
  CalculatorCaller(RpcCallerEndpoint endpoint)
    : super(calculatorService, endpoint);

  Future<SumResponse> sum(SumRequest request, {RpcContext? context}) =>
      callUnary<SumRequest, SumResponse>(
        methodName: 'sum',
        request: request,
        requestCodec: RpcCodec(SumRequest.fromJson),
        responseCodec: RpcCodec(SumResponse.fromJson),
        context: context,
      );
}

Run them over the in-memory transport:

Future<void> main() async {
  final (clientTransport, serverTransport) = RpcChannelTransport.memoryPair();

  final responder = RpcResponderEndpoint(transport: serverTransport);
  responder.registerServiceContract(CalculatorResponder());
  responder.start(); // without start() calls hang

  final caller = RpcCallerEndpoint(transport: clientTransport);
  final res = await CalculatorCaller(caller).sum(SumRequest([1, 2, 3]));
  print(res.result); // 6.0

  await caller.close();
  await responder.close();
}

Swap memoryPair() for a transport from a transport package and nothing else changes.

Use rpc_dart_generator to generate caller/responder boilerplate from annotated Dart interfaces.

---

Protobuf / binary codecs

RpcBinaryCodec adapts any byte format; the type does not need to implement IRpcSerializable. With protobuf-generated classes:

final reqCodec = RpcBinaryCodec<MyRequest>(
  toBytes: (r) => r.writeToBuffer(),
  fromBytes: MyRequest.fromBuffer,
);

---

Resilience

Retry and circuit breaker

Interceptors are added with addInterceptor; the first one added is the outermost. Add the circuit breaker before the retry so an open circuit fails fast instead of being retried:

void configureCaller(RpcCallerEndpoint caller) {
  caller
    ..addInterceptor(
      RpcCircuitBreakerInterceptor(
        failureThreshold: 5,
        resetTimeout: const Duration(seconds: 30),
      ),
    )
    ..addInterceptor(
      RpcRetryInterceptor(
        maxAttempts: 3,
        backoff: const RpcExponentialBackoff(
          baseDelay: Duration(milliseconds: 100),
          maxDelay: Duration(seconds: 5),
        ),
      ),
    );
}

Retry covers unary calls only and is meant for idempotent methods. RpcFixedBackoff is the other built-in RpcBackoffPolicy.

Client connection with reconnect

RpcClientConnection is the one layer that reconnects: a transport closes when its connection is lost, and the connection builds a new one from the factory. Create the endpoint once on connection.transport; it stays valid across reconnects:

RpcCallerEndpoint connect(Future<IRpcTransport> Function() open) {
  final connection = RpcClientConnection(
    transportFactory: (_) => open(),
    backoff: const RpcExponentialBackoff(maxDelay: Duration(seconds: 30)),
  );
  connection.connect(); // returns void; watch connection.state
  return RpcCallerEndpoint(transport: connection.transport);
}

Rate limiter

A responder-side interceptor:

void limit(RpcResponderEndpoint responder) {
  const second = Duration(seconds: 1);
  responder.addInterceptor(
    RpcRateLimitInterceptor(
      global: const RpcRateLimit.slidingWindow(max: 5000, window: second),
      perKeyFallback: const RpcRateLimit.slidingWindow(max: 100, window: second),
      keyExtractor: (call) => call.context.getHeader('user-id'),
    ),
  );
}

A refused call fails with RESOURCE_EXHAUSTED plus retry info, which the caller's default retry policy treats as transient.

---

gRPC Health Checking

GrpcHealthCheckContract implements the standard grpc.health.v1.Health protocol (Check and Watch), backed by a GrpcHealthServiceStatus you update. The service name '' means the whole server:

GrpcHealthServiceStatus serveHealth(RpcResponderEndpoint responder) {
  final status = GrpcHealthServiceStatus()
    ..setStatus('', GrpcServingStatus.serving)
    ..setStatus('MyService', GrpcServingStatus.serving);
  responder.registerServiceContract(GrpcHealthCheckContract(status));
  return status;
}

The messages are protobuf on the wire, so any standard gRPC health client (for example grpc_health_probe or a Kubernetes gRPC probe) can query it over rpc_dart_http2. Check on an unregistered service fails with NOT_FOUND.

---

Transport architecture

A 3-layer architecture:

IRpcChannel              — raw byte pipe (WebSocket, TCP, etc.)
IRpcMultiplexedChannel   — multiplexed framed messages
IRpcTransport            — full transport with stream IDs and health

An application imports package:rpc_dart/rpc_dart.dart. A transport author imports package:rpc_dart/transport.dart instead: it re-exports the application library and adds channels, frames, the message parser and router, the transport capability interfaces, the status trailers a transport builds (RpcTrailers), and the server's shared parts: RpcConnectionHost for one endpoint per connection, RpcWriteBacklog to stop reading a peer that reads nothing, RpcWriteCoalescer to join the frames of one event-loop turn into one write, and the drain helpers. A server on a dart:io socket also imports package:rpc_dart/transport_io.dart, whose RpcHeaderBoundedServerSocket bounds an HTTP request's header block in bytes and time. A client there gets RpcSecureSocket.startConnect, a TLS connect for HttpClient.connectionFactory whose cancel also closes a socket still in the handshake.

Factories on RpcChannelTransport:

import 'package:rpc_dart/transport.dart';

void transports(IRpcChannel myChannel) {
  // Zero-copy in-memory pair (tests, in-process):
  final (client, server) = RpcChannelTransport.memoryPair();

  // Frame-based pair (exercises the full codec path):
  final (frameClient, frameServer) = RpcChannelTransport.pair();

  // Wrap any IRpcChannel (WebSocket, TCP):
  final transport = RpcChannelTransport.fromChannel(
    channel: myChannel,
    isClient: true,
  );
}

memoryPair() passes objects by reference, so it skips codecs on unary calls and never enforces byte limits. If production uses a network transport, run your tests over pair() too.

---

Flow control and resource limits

Both are configured through RpcSecurityPolicy, which is passed to the transport (memoryPair(policy:), pair(policy:), fromChannel(policy:), or a transport package's policy parameter), not to the endpoint. The defaults are safe for a server exposed to untrusted peers; you only need to touch them to relax or tighten.

Flow control

A slow or absent consumer must not let a producer pin memory. Each stream has a window, and all streams on a connection share a second one:

final (client, server) = RpcChannelTransport.pair(
  policy: const RpcSecurityPolicy(
    flowControlWindowBytes: 4 * 1024 * 1024,            // per stream
    flowControlConnectionWindowBytes: 64 * 1024 * 1024, // whole connection
  ),
);

The sender blocks once its window is used up and resumes as the receiving application consumes. Credit travels on bare metadata frames; a peer that does not understand them ignores them, and a grace window keeps the sender from deadlocking against such a peer.

Set either to null to disable. HTTP/2 has its own flow control, so rpc_dart_http2 should disable these rather than run two windows over each other.

Resource limits

const policy = RpcSecurityPolicy(
  maxActiveStreams: 4096,          // concurrent streams, per connection
  maxConcurrentHandlers: 256,      // handlers RUNNING, default null
  maxMessageLengthBytes: 16 << 20, // single decoded message
  maxBufferedBytes: 16 << 20,      // queued for a stream nobody is reading
  maxMetadataBytes: 64 * 1024,
  maxHeaders: 128,
  halfOpenStreamTimeout: Duration(seconds: 60),
  closeOnProtocolError: false,     // the default; a 256-violation backstop
);                                 // closes the connection either way

maxActiveStreams applies to streams the peer opens, not just your own calls. halfOpenStreamTimeout reclaims a stream that was opened but never carried a request — set it to null to disable. Note it covers dispatch only: a stream that has reached a handler and then goes idle is deliberately not reclaimed, because that is also what a legitimate long-lived subscription looks like.

maxConcurrentHandlers is a different dimension from maxActiveStreams, which bounds stream STATE. Dart cannot interrupt a running async function, so a handler that ignores its cancellation token keeps going after its stream is reclaimed and its admission slot returned — a peer pacing its calls past the reclaim grace accumulates handlers with nothing to stop them. At the ceiling a new call is refused with RESOURCE_EXHAUSTED, which is retryable.

maxBufferedBytes bounds what is queued for a stream whose consumer is not reading. Metadata is deliberately exempt from flow control — HTTP/2 exempts HEADERS for the same reason, a control frame that cannot be sent deadlocks the stream it is trying to end — so size is what bounds it. Over the limit the stream fails with RESOURCE_EXHAUSTED and the connection survives.

---

Health monitoring

Every endpoint reports its own and its transport's health:

Future<void> checkHealth(RpcCallerEndpoint caller) async {
  final report = await caller.health();
  if (!report.isHealthy) {
    print('issue: ${report.transportStatus.message}');
  }
}

---

Cancellation and deadlines

Future<void> cancellable(CalculatorCaller calculator) async {
  final token = RpcCancellationToken();
  final ctx = RpcContext.empty().withCancellation(
    token,
  ).withTimeout(const Duration(seconds: 2));

  final pending = calculator.sum(SumRequest([1, 2]), context: ctx);
  token.cancel('user cancelled'); // the call fails with CANCELLED
  await pending.catchError((_) => SumResponse(0));
}

// In a handler, cooperate with cancellation:
Future<SumResponse> slowSum(SumRequest request, {RpcContext? context}) async {
  var total = 0.0;
  for (final value in request.values) {
    context?.cancellationToken?.throwIfCancelled();
    total += value;
  }
  return SumResponse(total);
}

A cancelled token poisons its context: make a new token or context per logical operation.

---

Error handling

Throw RpcStatusException with an RpcStatus code to send a status to the caller. Any other exception reaches the caller as INTERNAL.

Future<SumResponse> guardedSum(SumRequest request, {RpcContext? context}) async {
  if (context?.getHeader('authorization') == null) {
    throw RpcStatusException(RpcStatus.unauthenticated, 'Not authorized');
  }
  return SumResponse(request.values.fold<double>(0, (a, b) => a + b));
}

On the caller, branch on e.statusCode: a status sent by the server always arrives as a plain RpcStatusException.

---

Testing

import 'package:test/test.dart';

void main() {
  test('sum', () async {
    final (ct, st) = RpcChannelTransport.memoryPair();
    final responder = RpcResponderEndpoint(transport: st)
      ..registerServiceContract(CalculatorResponder())
      ..start();
    final caller = RpcCallerEndpoint(transport: ct);

    final res = await CalculatorCaller(caller).sum(SumRequest([1, 2, 3]));
    expect(res.result, 6.0);

    await caller.close();
    await responder.close();
  });
}

---

AI agents

This package ships an Agent Skill for coding agents (Claude Code, Cursor, Copilot, Codex and others): the API reference, rules and pitfalls, matched to the rpc_dart version your project resolves. Install it from your project root with the skills CLI:

dart run skills@ get

It is installed into your agent's skills directory, for example .claude/skills/ or .agents/skills/. If your SDK does not support dart run <package>@, add skills as a dev dependency and run dart run skills get.

---

Versioning

A minor release may add members to the classes you build on: IRpcTransport, IRpcChannel, IRpcMultiplexedChannel, IRpcCodec, IRpcInterceptor, IRpcServer, RpcCompressionCodec and RpcBackoffPolicy. A new member comes with a default body, so code that extends them keeps compiling. Code that implements them may have to add the member. Mocks that implement an interface are fine in tests; for production code, extend. A transport decorator implements IRpcTransport on purpose, so that it forwards every member; it is the one place to expect such an addition.

The contract base classes (RpcCallerContract, RpcResponderContract, RpcPeerContract) gain no public member in 7.x. A generated contract extends them, so each of their names is one a service method cannot have, and the generator refuses those names.

The small role interfaces never gain a member: IRpcSerializable, IRpcFlowCredit and IRpcConnectionBuffer. A new capability comes as a new interface.

A minor release may add a value to these enums or a subtype to these sealed families, so a switch over one of them keeps a default branch: RpcHealthLevel, RpcClientConnectionState, RpcCallOutcome, RpcDataTransferMode, RpcContentTypeValidation, ConsoleFormat, LogSpanStatus, GrpcServingStatus (it mirrors a protobuf enum, which can grow), RpcRateLimit and LogRecord. These are closed and stay as they are: RpcMethodType, RpcCallDirection, RpcCircuitBreakerState, RpcServiceKind, RpcServerMode, RpcLogLevel and RpcProtoWireType.

An error detail of a google.rpc type that arrives as RpcRawErrorDetail today may be decoded into its own type in a minor release. Code that reads a detail by typeUrl from a raw one also handles the typed class.

collectEndpointMetrics() returns a typed snapshot, one type per endpoint: RpcCallerEndpointMetrics, RpcResponderEndpointMetrics or RpcPeerEndpointMetrics. A minor release may add a field to them. The keys of a health report's details are diagnostics, not API: read them for display, and do not build on a key.

package:rpc_dart/transport.dart, the library for transport authors, is looser than the rest: what it alone exports (channels and frames, the message parser and router, the stream-id and write helpers) may change in a minor release, and the CHANGELOG says so when it does. package:rpc_dart/rpc_dart.dart does not change incompatibly in 7.x.

These defaults hold for all of 7.x, so tightening them is up to you:

• No cap on connections (maxConnections on a server) or on concurrently running handlers (RpcSecurityPolicy.maxConcurrentHandlers). Per-connection stream, buffer and message limits are on by default. A server open to the internet sets both caps for its own load.
• RpcSecurityPolicy.contentTypeValidation is lenient: a request without a content-type is served.
• RpcRetryInterceptor retries a unary call that ends UNAVAILABLE, whether or not the method is idempotent. A lost response is also UNAVAILABLE, so a retried call may run twice on the server. Attach the interceptor to an endpoint for idempotent methods only, pass retryOn, or deduplicate on RpcContext.requestId.

Every metadata name starting with x-rpc- is reserved (RpcHeaders.reservedPrefix), so a minor release can add a control header without taking a name your metadata uses.

The channel transports' wire format carries a version (see RpcChannelFrame.protocolVersion). The client states it when it connects and the server answers in the same version, so a server can go on serving older clients; a version the other side does not speak fails the calls with FAILED_PRECONDITION instead of misreading frames. Upgrade servers first.

Ecosystem

This package is the core of the rpc_dart ecosystem. Additional packages are available:

Package	Description
rpc_dart_generator	Code generator for callers and responders
rpc_dart_framework	Server framework: modules, DI, lifecycle, rate limiting
rpc_dart_grpc_reflection	gRPC Server Reflection (grpcurl, Postman support)
rpc_dart_opentelemetry	OpenTelemetry tracing and metrics
rpc_dart_http2	HTTP/2 transport (gRPC wire compatible)
rpc_dart_http	HTTP/1.1 transport
rpc_dart_websocket	WebSocket transport
rpc_dart_isolate	Isolate transport
rpc_dart_wasm	WebAssembly bridge transport

For the full list visit the GitHub repository.