AcpServerConnectionState constructor

AcpServerConnectionState({
  1. required String connectionId,
  2. required AcpWritable<Object?> inbound,
  3. required Stream<Object?> outbound,
  4. int outboundCapacity = 1024,
  5. int maximumSessions = 10000,
  6. int maximumPendingRoutes = 4096,
  7. bool allowBatches = false,
  8. AcpOutboundOverflowHandler? onOverflow,
})

Creates and starts a server connection state.

Implementation

AcpServerConnectionState({
  required this.connectionId,
  required AcpWritable<Object?> inbound,
  required Stream<Object?> outbound,
  this.outboundCapacity = 1024,
  this.maximumSessions = 10000,
  this.maximumPendingRoutes = 4096,
  bool allowBatches = false,
  AcpOutboundOverflowHandler? onOverflow,
}) : _inbound = inbound,
     _allowBatches = allowBatches,
     allOutbound = AcpOutboundHub<Object?>(
       capacity: outboundCapacity,
       onOverflow: onOverflow,
     ),
     connectionOutbound = AcpOutboundHub<Object?>(
       capacity: outboundCapacity,
       onOverflow: onOverflow,
     ),
     _onOverflow = onOverflow {
  if (connectionId.isEmpty) {
    throw ArgumentError.value(
      connectionId,
      'connectionId',
      'must not be empty',
    );
  }
  if (maximumSessions <= 0) {
    throw ArgumentError.value(
      maximumSessions,
      'maximumSessions',
      'must be positive',
    );
  }
  if (maximumPendingRoutes <= 0) {
    throw ArgumentError.value(
      maximumPendingRoutes,
      'maximumPendingRoutes',
      'must be positive',
    );
  }
  // The state owns this subscription and cancels it in close().
  // ignore: cancel_subscriptions
  final StreamSubscription<Object?> subscription = outbound.listen(
    (Object? frame) {
      try {
        _routeOutbound(frame);
      } on Object catch (error) {
        unawaited(close(error));
      }
    },
    onError: (Object error, StackTrace stackTrace) {
      unawaited(close(error));
    },
    onDone: close,
  );
  _outboundSubscription = subscription;
}