23169302222

Committed 16 Mar 2026 10:40PM UTC coverage: 76.365% (+0.2%) from 76.156%

Build # 23169302222

Build Type

push

github

Committed by

nogipx

Commit Message

update deps

Coverage Stats

3887 of 5090 relevant lines covered (76.37%)

8.09 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.75

/packages/rpc_dart/lib/src/rpc/streams/base_processor.dart

// SPDX-FileCopyrightText: 2025 Karim "nogipx" Mamatkazin <nogipx@gmail.com>
// SPDX-FileCopyrightText: 2026 Karim "nogipx" Mamatkazin <nogipx@gmail.com>
//
// SPDX-License-Identifier: MIT

part of '_index.dart';

/// Shared stream processor: zero-copy when no codecs (zero-copy transport required), otherwise serialized.
final class StreamProcessor<TRequest extends Object, TResponse extends Object> {
  final RpcLogger? _logger;
  final IRpcTransport _transport;
  final int _streamId;
  final String _serviceName;
  final String _methodName;
  final IRpcCodec<TRequest>? _requestCodec;
  final IRpcCodec<TResponse>? _responseCodec;

  /// RPC context with cancellation/metadata.
  final RpcContext? _context;

  /// Cancellation subscription.
  StreamSubscription? _cancellationSubscription;

  /// Parser for fragmented messages (serialization mode only).
  RpcMessageParser? _parser;

  /// Whether zero-copy mode is active.
  final bool _isZeroCopy;

  /// Incoming requests controller.
  final StreamController<TRequest> _requestController =
      StreamController<TRequest>();

  /// Outgoing responses controller.
  final StreamController<TResponse> _responseController =
      StreamController<TResponse>(sync: true);

  /// Subscription to outgoing responses.
  StreamSubscription<TResponse>? _responseSubscription;

  /// Subscription to incoming transport messages.
  StreamSubscription? _messageSubscription;

  /// Send sequence to preserve order and await completion before trailers.
  Future<void> _sendSequence = Future<void>.value();

  bool _trailerSent = false;

  /// Processor active flag.
  bool _isActive = true;

  /// Indicates initial metadata sent.
  bool _initialMetadataSent = false;

  /// Response encoding selected from client's grpc-accept-encoding.
  /// Null means identity (no compression).
  /// Initially set from server context; overridden by incoming client metadata.
  String? _responseEncoding;

  /// Request encoding advertised by the peer in grpc-encoding.
  /// Set when the initial request metadata arrives; used by the decompressor.
  String? _requestEncoding;

  /// Method path `/Service/Method`.
  late final String _methodPath;

  StreamProcessor({
    required IRpcTransport transport,
    required int streamId,
    required String serviceName,
    required String methodName,
    IRpcCodec<TRequest>? requestCodec,
    IRpcCodec<TResponse>? responseCodec,
    RpcContext? context,
    RpcLogger? logger,
  })  : _transport = transport,
        _streamId = streamId,
        _serviceName = serviceName,
        _methodName = methodName,
        _isZeroCopy = requestCodec == null && responseCodec == null,
        _requestCodec = requestCodec,
        _responseCodec = responseCodec,
        _context = context,
        _logger = logger?.child('StreamProcessor') {
    // Serialization requires codecs.
    if (!_isZeroCopy) {
      if (_requestCodec == null || _responseCodec == null) {
        throw ArgumentError(
          'Codecs are required for serialization mode. '
          'For zero-copy leave codecs null.',
        );
      }
      _parser = RpcMessageParser(
        logger: _logger,
        decompressor: (payload) {
          final encoding =
              _requestEncoding ?? _context?.getHeader(RpcHeaders.grpcEncoding);
          if (encoding == null || encoding == RpcGrpcCompression.identity) {
            throw RpcException(
              'Compressed gRPC payload received without grpc-encoding',
            );
          }
          return RpcGrpcCompression.decompress(payload, encoding: encoding);
        },
      );
    } else {
      // Zero-copy requires transport support.
      if (!transport.supportsZeroCopy) {
        throw ArgumentError(
          'Zero-copy mode requires a transport with zero-copy support. '
          'Provide codecs for network transports.',
        );
      }
    }

    _methodPath = '/$_serviceName/$_methodName';
    _responseEncoding = _pickResponseEncoding(context);

    _logger?.internal(
      'Created ${_isZeroCopy ? "Zero-copy" : "Serialized"} StreamProcessor for $_methodPath [streamId: $_streamId]${_context?.cancellationToken != null ? " with cancellation token" : ""}',
    );

    _setupCancellationMonitoring();
    _setupResponseHandler();
  }

  /// Picks the best response encoding from the client's grpc-accept-encoding.
  static String? _pickResponseEncoding(RpcContext? context) {
    final accept = context?.getHeader(RpcHeaders.grpcAcceptEncoding);
    return RpcGrpcCompression.selectResponseEncoding(accept);
  }

  /// Incoming request stream.
  Stream<TRequest> get requests => _requestController.stream;

  /// Whether processor is active.
  bool get isActive => _isActive;

  /// Zero-copy mode flag.
  bool get isZeroCopy => _isZeroCopy;

  /// Configures outgoing response handling.
  void _setupResponseHandler() {
    _responseSubscription = _responseController.stream.listen(
      (response) {
        _sendSequence = _sendSequence.then((_) async {
          if (!_isActive) return;

          _logger?.internal(
            'Sending response for $_methodPath [streamId: $_streamId]',
          );
          try {
            if (_isZeroCopy) {
              // Zero-copy path
              _logger?.internal(
                'Zero-copy send [streamId: $_streamId]',
              );
              await _transport.sendDirectObject(_streamId, response);
              _logger?.internal(
                'Zero-copy response sent for $_methodPath [streamId: $_streamId]',
              );
            } else {
              // Send initial metadata before the first response frame only when
              // we need to advertise compression. Without compression the
              // existing behaviour (no initial metadata for streaming) is kept
              // so existing tests and in-memory transports are not affected.
              if (_responseEncoding != null && !_initialMetadataSent) {
                await _transport.sendMetadata(
                  _streamId,
                  RpcMetadata.forServerInitialResponse(
                    encoding: _responseEncoding,
                  ),
                );
                _initialMetadataSent = true;
              }

              // Serialization for network transports
              final serialized = _responseCodec!.serialize(response);
              _logger?.internal(
                'Response serialized (${serialized.length} bytes) [streamId: $_streamId]',
              );

              final useCompression = _responseEncoding != null;
              final payload = useCompression
                  ? RpcGrpcCompression.compress(
                      serialized,
                      encoding: _responseEncoding!,
                    )
                  : serialized;
              final framedMessage = RpcMessageFrame.encode(
                payload,
                compressed: useCompression,
              );
              await _transport.sendMessage(_streamId, framedMessage);

              _logger?.internal(
                'Response sent for $_methodPath [streamId: $_streamId]',
              );
            }
          } catch (e, stackTrace) {
            // Skip if transport closed
            if (e.toString().contains('Transport is closed') ||
                e.toString().contains('closed')) {
              _logger?.internal(
                'Transport closed, skipping response send [streamId: $_streamId]',
              );
              return;
            }
            _logger?.error(
              'Failed to send response [streamId: $_streamId]',
              error: e,
              stackTrace: stackTrace,
            );
          }
        });
      },
      onError: (error, stackTrace) {
        _logger?.error(
          'Error in response stream for $_methodPath [streamId: $_streamId]',
          error: error,
          stackTrace: stackTrace,
        );
      },
    );
  }

  Future<void> _sendOkTrailerIfNeeded() async {
    if (_trailerSent) return;
    _trailerSent = true;

    try {
      final trailers = RpcMetadata.forTrailer(RpcStatus.ok);
      await _transport.sendMetadata(_streamId, trailers, endStream: true);
      _logger?.internal(
        'Trailer sent for $_methodPath [streamId: $_streamId]',
      );
    } catch (e, stackTrace) {
      if (e.toString().contains('Transport is closed') ||
          e.toString().contains('closed')) {
        _logger?.internal(
          'Transport closed, skipping trailer send [streamId: $_streamId]',
        );
        return;
      }
      _logger?.error(
        'Failed to send trailer [streamId: $_streamId]',
        error: e,
        stackTrace: stackTrace,
      );
    }
  }

  /// Binds the processor to an endpoint message stream.
  void bindToMessageStream(Stream<RpcTransportMessage> messageStream) {
    if (_messageSubscription != null) {
      _logger?.logRpcWarning(
        message: 'Stream processor already bound to message stream',
        methodPath: _methodPath,
        streamId: _streamId,
      );
      return;
    }

    _logger?.logStreamBound(methodPath: _methodPath, streamId: _streamId);

    _messageSubscription = messageStream.listen(
      _handleMessage,
      onError: (error, stackTrace) {
        _logger?.logRpcError(
          operation: 'message_stream_listen',
          error: error,
          stackTrace: stackTrace,
          methodPath: _methodPath,
          streamId: _streamId,
        );
        if (!_requestController.isClosed) {
          _requestController.addError(error, stackTrace);
        }
      },
      onDone: () {
        _logger?.logStreamFinished(
          methodPath: _methodPath,
          streamId: _streamId,
          reason: 'message_stream_completed',
        );
        if (!_requestController.isClosed) {
          _requestController.close();
        }
      },
    );

    // Initial metadata is not sent on bind; it is sent with the first response
    // or skipped when sending an immediate error.
  }

  /// Checks cancellation token and throws if cancelled.
  void _checkCancellation() {
    _context?.cancellationToken?.throwIfCancelled();
  }

  /// Handles an incoming transport message.
  void _handleMessage(RpcTransportMessage message) {
    if (!_isActive) return;

    // Check cancellation before handling each message.
    try {
      _checkCancellation();
    } catch (e) {
      _logger?.internal(
        'Message skipped due to cancellation [streamId: $_streamId]',
      );
      return;
    }

    _logger?.logMessageReceived(
      streamId: message.streamId,
      messageType: message.isMetadataOnly
          ? 'metadata'
          : message.isDirect
              ? 'zero_copy'
              : 'serialized',
      payloadSize: message.payload?.length,
      isDirectPayload: message.isDirect,
    );

    // Extract encoding hints from initial request metadata.
    if (message.isMetadataOnly && message.metadata != null) {
      final meta = message.metadata!;

      // grpc-encoding: what the peer used to compress its requests.
      final reqEnc = meta.getHeaderValue(RpcHeaders.grpcEncoding);
      if (reqEnc != null && reqEnc != RpcGrpcCompression.identity) {
        _requestEncoding = reqEnc;
      }

      // grpc-accept-encoding: what the peer can decompress → use for responses.
      if (_responseEncoding == null) {
        final accept = meta.getHeaderValue(
          RpcHeaders.grpcAcceptEncoding,
        );
        if (accept != null) {
          for (final enc in accept.split(',').map((e) => e.trim())) {
            if (enc != RpcGrpcCompression.identity &&
                RpcGrpcCompression.isSupported(enc)) {
              _responseEncoding = enc;
              break;
            }
          }
        }
      }
    }

    // Zero-copy: direct object.
    if (message.isDirect && message.directPayload != null) {
      _processDirectMessage(message.directPayload!);
    }
    // Serialized payload.
    else if (!message.isMetadataOnly && message.payload != null) {
      _processDataMessage(message.payload!);
    }

    // End of stream.
    if (message.isEndOfStream) {
      _logger?.logStreamFinished(
        methodPath: _methodPath,
        streamId: _streamId,
        reason: 'end_of_stream_received',
      );
      if (!_requestController.isClosed) {
        _requestController.close();
      }
    }
  }

  /// Zero-copy: processes a direct object without serialization.
  void _processDirectMessage(Object directPayload) {
    try {
      final request = directPayload as TRequest;

      if (!_requestController.isClosed) {
        _requestController.add(request);
      } else {
        _logger?.logRpcWarning(
          message: 'Cannot add request to closed controller (zero-copy)',
          methodPath: _methodPath,
          streamId: _streamId,
          metadata: {'transport_type': 'zero_copy'},
        );
      }
    } catch (e, stackTrace) {
      _logger?.logRpcError(
        operation: 'zero_copy_direct_object_processing',
        error: e,
        stackTrace: stackTrace,
        methodPath: _methodPath,
        streamId: _streamId,
        metadata: {'object_type': directPayload.runtimeType.toString()},
      );
      if (!_requestController.isClosed) {
        _requestController.addError(e, stackTrace);
      }
    }
  }

  /// Processes a serialized message (serialization mode only).
  void _processDataMessage(List<int> messageBytes) {
    if (_isZeroCopy) {
      _logger?.logRpcWarning(
        message: 'Serialized message received in zero-copy mode, ignoring',
        methodPath: _methodPath,
        streamId: _streamId,
      );
      return;
    }

    _logger?.logMessageReceived(
      streamId: _streamId,
      messageType: 'serialized_data',
      payloadSize: messageBytes.length,
    );

    try {
      // Convert to Uint8List for parser.
      final uint8Message = messageBytes is Uint8List
          ? messageBytes
          : Uint8List.fromList(messageBytes);

      final messages = _parser!(uint8Message);

      for (var msgBytes in messages) {
        try {
          final request = _requestCodec!.deserialize(msgBytes);

          if (!_requestController.isClosed) {
            _requestController.add(request);
          } else {
            _logger?.logRpcWarning(
              message: 'Cannot add request to closed controller',
              methodPath: _methodPath,
              streamId: _streamId,
              metadata: {'message_size': msgBytes.length},
            );
          }
        } catch (e, stackTrace) {
          _logger?.logRpcError(
            operation: 'request_deserialization',
            error: e,
            stackTrace: stackTrace,
            methodPath: _methodPath,
            streamId: _streamId,
            metadata: {'message_size': msgBytes.length},
          );
          if (!_requestController.isClosed) {
            _requestController.addError(e, stackTrace);
          }
        }
      }
    } catch (e, stackTrace) {
      _logger?.logRpcError(
        operation: 'message_parsing',
        error: e,
        stackTrace: stackTrace,
        methodPath: _methodPath,
        streamId: _streamId,
        metadata: {'message_size': messageBytes.length},
      );
      if (!_requestController.isClosed) {
        _requestController.addError(e, stackTrace);
      }
    }
  }

  /// Sends a response to the client.
  Future<void> send(TResponse response) async {
    if (!_isActive) {
      _logger?.warning('Attempted to send response on inactive processor');
      return;
    }

    // Check cancellation before sending a response.
    try {
      _checkCancellation();
    } catch (e) {
      _logger?.internal(
        'Response skipped due to cancellation [streamId: $_streamId]',
      );
      return;
    }

    if (!_responseController.isClosed) {
      _responseController.add(response);
    } else {
      _logger?.warning('Attempted to send response to closed controller');
    }
  }

  /// Sends an error to the client.
  Future<void> sendError(int statusCode, String message) async {
    if (!_isActive) {
      _logger?.warning('Attempted to send error on inactive processor');
      return;
    }

    _logger?.error(
      'Sending error to client: $statusCode - $message [streamId: $_streamId]',
    );

    // Wait for pending sends to avoid interleaving the error trailer.
    await _sendSequence;

    if (!_responseController.isClosed) {
      await _responseController.close();
    }

    try {
      // If initial metadata was not sent, send an error response immediately.
      if (!_initialMetadataSent) {
        _logger?.internal(
          'Sending error without initial metadata [streamId: $_streamId]',
        );
        // Create combined metadata: initial response + error trailer.
        final errorHeaders = [
          RpcHeader(':status', '200'), // HTTP 200 for gRPC
          RpcHeader(
            RpcHeaders.contentType,
            RpcHeaders.contentTypeGrpc,
          ),
          RpcHeader(RpcHeaders.grpcStatus, statusCode.toString()),
        ];

        if (message.isNotEmpty) {
          errorHeaders.add(
            RpcHeader(
              RpcHeaders.grpcMessage,
              RpcMetadata.encodeGrpcMessage(message),
            ),
          );
        }

        final errorMetadata = RpcMetadata(errorHeaders);
        await _transport.sendMetadata(
          _streamId,
          errorMetadata,
          endStream: true,
        );
        _initialMetadataSent = true;
      } else {
        // Initial metadata already sent; send trailer only.
        final trailers = RpcMetadata.forTrailer(statusCode, message: message);
        await _transport.sendMetadata(_streamId, trailers, endStream: true);
      }

      _logger?.internal('Error sent to client [streamId: $_streamId]');
      _trailerSent = true;
    } catch (e, stackTrace) {
      // Skip if transport is closed.
      if (e.toString().contains('Transport is closed') ||
          e.toString().contains('closed')) {
        _logger?.internal(
          'Transport closed, skipping error send [streamId: $_streamId]',
        );
        return;
      }
      _logger?.error(
        'Failed to send error to client [streamId: $_streamId]',
        error: e,
        stackTrace: stackTrace,
      );
    }
  }

  /// Finishes sending responses.
  Future<void> finishSending() async {
    if (!_isActive) return;

    _logger?.internal(
      'Finishing response send for $_methodPath [streamId: $_streamId]',
    );

    await _sendSequence;

    if (!_responseController.isClosed) {
      await _responseController.close();
    }

    await _sendOkTrailerIfNeeded();
  }

  /// Closes the processor and frees resources.
  Future<void> close() async {
    if (!_isActive) return;

    _logger?.internal(
      'Closing StreamProcessor for $_methodPath [streamId: $_streamId]',
    );
    _isActive = false;

    // Cancel all subscriptions.
    unawaited(_messageSubscription?.cancel());
    _messageSubscription = null;

    unawaited(_cancellationSubscription?.cancel());
    _cancellationSubscription = null;

    unawaited(_responseSubscription?.cancel());
    _responseSubscription = null;

    if (!_requestController.isClosed) {
      _requestController.close();
    }

    if (!_responseController.isClosed) {
      _responseController.close();
    }
  }

  /// Sets up cancellation monitoring.
  void _setupCancellationMonitoring() {
    if (_context?.cancellationToken != null) {
      _cancellationSubscription =
          _context!.cancellationToken!.cancelled.asStream().listen(
        (_) {
          _logger?.internal(
            'Operation cancelled, shutting down processor [streamId: $_streamId]',
          );
          _isActive = false;

          final reason =
              _context!.cancellationToken!.reason ?? 'Operation was cancelled';
          final cancelledException = RpcCancelledException(reason);

          if (!_requestController.isClosed) {
            _requestController.addError(cancelledException);
          }
          if (!_responseController.isClosed) {
            _responseController.addError(cancelledException);
          }

          // Cancel subscriptions.
          _messageSubscription?.cancel();
          _cancellationSubscription?.cancel();
        },
        onError: (error, stackTrace) {
          _logger?.error(
            'Error monitoring cancellation [streamId: $_streamId]',
            error: error,
            stackTrace: stackTrace,
          );
        },
      );
    }
  }
}

/// Shared processor for client RPC stream calls.
///
/// Automatically selects mode:
/// - Zero-copy for in-memory transport (codecs not needed)
/// - Serialization for network transports (codecs required)
///
/// Benefits:
/// - Reuse across stream call types
/// - Avoids race conditions
/// - Clear separation of concerns
/// - Testable without out-of-process dependencies
/// - Works with any object types, not just IRpcSerializable
/// - Auto-optimized for in-memory transport
final class CallProcessor<TRequest extends Object, TResponse extends Object> {
  final RpcLogger? _logger;
  final IRpcTransport _transport;
  final int _streamId;
  final String _serviceName;
  final String _methodName;
  final IRpcCodec<TRequest>? _requestCodec;
  final IRpcCodec<TResponse>? _responseCodec;

  /// RPC context for metadata, timeouts, and cancellation.
  final RpcContext? _context;

  /// Cancellation subscription.
  StreamSubscription? _cancellationSubscription;

  /// Parser for fragmented messages (serialization mode only).
  RpcMessageParser? _parser;

  String? _peerGrpcEncoding;

  /// Processor mode flag.
  final bool _isZeroCopy;

  /// Outgoing request controller.
  final StreamController<TRequest> _requestController =
      StreamController<TRequest>();

  /// Incoming response controller.
  final StreamController<RpcMessage<TResponse>> _responseController =
      StreamController<RpcMessage<TResponse>>();

  /// Outgoing request subscription.
  StreamSubscription? _requestSubscription;

  /// Incoming response subscription.
  StreamSubscription? _responseSubscription;

  /// Processor active flag.
  bool _isActive = true;

  /// Whether initial metadata was sent.
  bool _initialMetadataSent = false;

  /// Method path in /Service/Method format.
  late final String _methodPath;

  CallProcessor({
    required IRpcTransport transport,
    required String serviceName,
    required String methodName,
    IRpcCodec<TRequest>? requestCodec,
    IRpcCodec<TResponse>? responseCodec,
    RpcContext? context,
    RpcLogger? logger,
  })  : _transport = transport,
        _streamId = transport.createStream(),
        _serviceName = serviceName,
        _methodName = methodName,
        _isZeroCopy = requestCodec == null && responseCodec == null,
        _requestCodec = requestCodec,
        _responseCodec = responseCodec,
        _context = context,
        _logger = logger?.child('CallProcessor') {
    // Validation: codecs are required for serialization mode.
    if (!_isZeroCopy) {
      if (_requestCodec == null || _responseCodec == null) {
        throw ArgumentError(
          'Codecs are required for serialization mode. '
          'For zero-copy leave codecs null.',
        );
      }
      _parser = RpcMessageParser(
        logger: _logger,
        decompressor: (payload) {
          final encoding = _peerGrpcEncoding;
          if (encoding == null || encoding == RpcGrpcCompression.identity) {
            throw RpcException(
              'Compressed gRPC payload received without grpc-encoding',
            );
          }
          return RpcGrpcCompression.decompress(payload, encoding: encoding);
        },
      );
    } else {
      // Zero-copy mode requires transport support.
      if (!transport.supportsZeroCopy) {
        throw ArgumentError(
          'Zero-copy mode requires a transport with zero-copy support. '
          'Provide codecs for network transports.',
        );
      }
    }

    _methodPath = '/$_serviceName/$_methodName';

    _logger?.internal(
      'Created ${_isZeroCopy ? "Zero-copy" : "Serialized"} CallProcessor for $_methodPath [streamId: $_streamId]${_context?.cancellationToken != null ? " with cancellation token" : ""}',
    );

    // Validate context before starting.
    _checkContextBeforeCall();

    _setupCancellationMonitoring();
    _setupRequestHandler();
    _setupResponseHandler();
  }

  /// Incoming responses from server.
  Stream<RpcMessage<TResponse>> get responses => _responseController.stream;

  /// Whether processor is active.
  bool get isActive => _isActive;

  /// Stream ID.
  int get streamId => _streamId;

  /// Zero-copy mode flag.
  bool get isZeroCopy => _isZeroCopy;

  /// Configures outgoing request handling.
  void _setupRequestHandler() {
    _requestSubscription = _requestController.stream.listen(
      (request) async {
        if (!_isActive) return;

        try {
          // Send initial metadata with the first request.
          if (!_initialMetadataSent) {
            await _sendInitialMetadata();
            _initialMetadataSent = true;
          }

          _logger?.internal(
            'Sending request for $_methodPath [streamId: $_streamId]',
          );

          if (_isZeroCopy) {
            // Zero-copy path.
            _logger?.internal(
              'Zero-copy request send [streamId: $_streamId]',
            );
            await _transport.sendDirectObject(_streamId, request);
            _logger?.internal(
              'Zero-copy request sent for $_methodPath [streamId: $_streamId]',
            );
          } else {
            // Serialization for network transports.
            final serialized = _requestCodec!.serialize(request);
            _logger?.internal(
              'Request serialized (${serialized.length} bytes) [streamId: $_streamId]',
            );

            final requestEncoding =
                _context?.getHeader(RpcHeaders.grpcEncoding);
            if (requestEncoding != null &&
                requestEncoding != RpcGrpcCompression.identity &&
                !RpcGrpcCompression.isSupported(requestEncoding)) {
              throw RpcException(
                'Unsupported grpc-encoding: $requestEncoding. '
                'Supported: ${RpcGrpcCompression.supportedEncodings().join(', ')}',
              );
            }
            final useCompression = requestEncoding != null &&
                requestEncoding != RpcGrpcCompression.identity;
            final payload = useCompression
                ? RpcGrpcCompression.compress(
                    serialized,
                    encoding: requestEncoding,
                  )
                : serialized;

            final framedMessage = RpcMessageFrame.encode(
              payload,
              compressed: useCompression,
            );
            await _transport.sendMessage(_streamId, framedMessage);

            _logger?.internal(
              'Request sent for $_methodPath [streamId: $_streamId]',
            );
          }
        } catch (e, stackTrace) {
          _logger?.error(
            'Failed to send request [streamId: $_streamId]',
            error: e,
            stackTrace: stackTrace,
          );
          if (!_responseController.isClosed) {
            _responseController.addError(e, stackTrace);
          }

          // Critical: on routing error stop immediately to prevent further sends.
          if (!_requestController.isClosed) {
            _requestController.close();
          }
        }
      },
      onDone: () async {
        if (!_isActive) return;

        try {
          await _transport.finishSending(_streamId);
          _logger?.internal(
            'finishSending completed for $_methodPath [streamId: $_streamId]',
          );
        } catch (e, stackTrace) {
          _logger?.error(
            'Failed to finish sending requests [streamId: $_streamId]',
            error: e,
            stackTrace: stackTrace,
          );
        }
      },
      onError: (error, stackTrace) {
        _logger?.error(
          'Error in request stream for $_methodPath [streamId: $_streamId]',
          error: error,
          stackTrace: stackTrace,
        );
        if (!_responseController.isClosed) {
          _responseController.addError(error, stackTrace);
        }
      },
    );
  }

  /// Configures incoming response handling.
  void _setupResponseHandler() {
    _responseSubscription = _transport.getMessagesForStream(_streamId).listen(
      _handleResponse,
      onError: (error, stackTrace) {
        _logger?.error(
          'Error in response stream',
          error: error,
          stackTrace: stackTrace,
        );
        if (!_responseController.isClosed) {
          _responseController.addError(error, stackTrace);
        }
      },
      onDone: () {
        _logger?.internal(
          'Response stream completed for $_methodPath [streamId: $_streamId]',
        );
        if (!_responseController.isClosed) {
          _responseController.close();
        }
      },
    );
  }

  /// Sends initial metadata with context support.
  Future<void> _sendInitialMetadata() async {
    _logger?.internal(
      'Sending initial metadata for $_methodPath [streamId: $_streamId]',
    );

    final baseMetadata =
        RpcMetadata.forClientRequest(_serviceName, _methodName);

    // Use a map so context headers naturally override base headers,
    // preventing duplicates (e.g. grpc-accept-encoding).
    final headerMap = <String, String>{
      for (final h in baseMetadata.headers) h.name: h.value,
    };

    if (_context != null) {
      // Context headers override base — handles grpc-accept-encoding dedup.
      headerMap.addAll(_context!.headers);

      if (_context!.traceId != null) {
        headerMap[RpcHeaders.xTraceId] = _context!.traceId!;
      }
      headerMap[RpcHeaders.xRequestId] = _context!.requestId;

      if (_context!.deadline != null) {
        final timeout = _context!.remainingTime;
        if (timeout != null) {
          headerMap[RpcHeaders.grpcTimeout] =
              RpcMetadata.encodeGrpcTimeout(timeout);
        }
      }

      _logger?.internal(
        'Context headers added: ${_context!.headers.length} custom + system [streamId: $_streamId]',
      );
    } else {
      headerMap[RpcHeaders.xRequestId] = RpcContext.empty().requestId;

      _logger?.internal(
        'Added base request-id for null context [streamId: $_streamId]',
      );
    }

    final metadata = RpcMetadata(
      [for (final e in headerMap.entries) RpcHeader(e.key, e.value)],
      methodPath: baseMetadata.methodPath,
    );
    await _transport.sendMetadata(_streamId, metadata);

    _logger?.internal(
      'Initial metadata sent for $_methodPath [streamId: $_streamId]',
    );
  }

  /// Sets up cancellation monitoring for CallProcessor.
  void _setupCancellationMonitoring() {
    if (_context?.cancellationToken != null) {
      _cancellationSubscription =
          _context!.cancellationToken!.cancelled.asStream().listen(
        (_) async {
          _logger?.internal(
            'Operation cancelled by client, notifying server [streamId: $_streamId]',
          );

          try {
            // Send a cancellation notice to the server.
            final reason = _context!.cancellationToken!.reason ??
                'Operation cancelled by client';
            await _sendCancellationToServer(reason);
          } catch (e, stackTrace) {
            _logger?.error(
              'Failed to send cancellation notice [streamId: $_streamId]',
              error: e,
              stackTrace: stackTrace,
            );
          }

          _isActive = false;
          final cancelledException = RpcCancelledException(
            _context!.cancellationToken!.reason ?? 'Operation was cancelled',
          );

          if (!_requestController.isClosed) {
            _requestController.addError(cancelledException);
          }
          if (!_responseController.isClosed) {
            _responseController.addError(cancelledException);
          }

          // Cancel subscriptions.
          await _requestSubscription?.cancel();
          await _responseSubscription?.cancel();
          _cancellationSubscription?.cancel();
        },
        onError: (error, stackTrace) {
          _logger?.error(
            'Error monitoring cancellation [streamId: $_streamId]',
            error: error,
            stackTrace: stackTrace,
          );
        },
      );
    }
  }

  /// Sends a cancellation notice to the server.
  Future<void> _sendCancellationToServer(String reason) async {
    try {
      // Build metadata with cancellation details.
      final cancellationHeaders = [
        RpcHeader('x-client-cancelled', 'true'),
        RpcHeader('x-cancellation-reason', reason),
        RpcHeader(
          RpcHeaders.grpcStatus,
          RpcStatus.cancelled.toString(),
        ),
      ];

      final cancellationMetadata = RpcMetadata(cancellationHeaders);

      _logger?.internal(
        'Sending cancellation notice to server [streamId: $_streamId]',
      );

      await _transport.sendMetadata(
        _streamId,
        cancellationMetadata,
        endStream: true,
      );

      _logger?.internal(
        'Cancellation notice sent to server [streamId: $_streamId]',
      );
    } catch (e, stackTrace) {
      _logger?.error(
        'Failed to send cancellation metadata [streamId: $_streamId]',
        error: e,
        stackTrace: stackTrace,
      );
    }
  }

  /// Validates context before making the call.
  void _checkContextBeforeCall() {
    if (_context == null) return;

    // Check cancellation.
    _context!.cancellationToken?.throwIfCancelled();

    // Check deadline.
    if (_context!.isExpired) {
      throw RpcDeadlineExceededException(_context!.deadline!, Duration.zero);
    }

    _logger?.internal(
      'Context verified: requestId=${_context!.requestId}, traceId=${_context!.traceId} [streamId: $_streamId]',
    );
  }

  /// Handles an incoming response.
  void _handleResponse(RpcTransportMessage message) {
    if (!_isActive) return;

    _logger?.internal(
      'Handling response [streamId: ${message.streamId}, isMetadataOnly: ${message.isMetadataOnly}, hasPayload: ${message.payload != null}, isDirect: ${message.isDirect}]',
    );

    try {
      // Handle metadata.
      if (message.isMetadataOnly) {
        final encoding = message.metadata?.getHeaderValue(
          RpcHeaders.grpcEncoding,
        );
        if (encoding != null) {
          _peerGrpcEncoding = encoding;
        }

        final rpcMessage = RpcMessage.withMetadata<TResponse>(
          message.metadata!,
          isEndOfStream: message.isEndOfStream,
        );

        if (!_responseController.isClosed) {
          _responseController.add(rpcMessage);
          _logger?.internal(
            'Metadata pushed to response stream [streamId: $_streamId]',
          );
        }
      }

      // Zero-copy: process direct object.
      if (message.isDirect && message.directPayload != null) {
        _processDirectResponse(message.directPayload!);
      }
      // Handle serialized payload.
      else if (!message.isMetadataOnly && message.payload != null) {
        _processResponseData(message.payload!);
      }

      // Finish stream on END_STREAM.
      if (message.isEndOfStream) {
        _logger?.internal(
          'END_STREAM received, closing response stream [streamId: $_streamId]',
        );
        if (!_responseController.isClosed) {
          _responseController.close();
        }
      }
    } catch (e, stackTrace) {
      _logger?.error(
        'Failed to process response [streamId: $_streamId]',
        error: e,
        stackTrace: stackTrace,
      );
      if (!_responseController.isClosed) {
        _responseController.addError(e, stackTrace);
      }
    }
  }

  /// Zero-copy: handles a direct response object without serialization.
  void _processDirectResponse(Object directPayload) {
    _logger?.internal(
      'Zero-copy response handling [streamId: $_streamId, type: ${directPayload.runtimeType}]',
    );

    try {
      final response = directPayload as TResponse;
      final rpcMessage = RpcMessage.withPayload<TResponse>(response);

      if (!_responseController.isClosed) {
        _responseController.add(rpcMessage);
        _logger?.internal(
          'Zero-copy response added to response stream [streamId: $_streamId]',
        );
      } else {
        _logger?.warning(
          'Zero-copy: cannot add response to closed controller [streamId: $_streamId]',
        );
      }
    } catch (e, stackTrace) {
      _logger?.error(
        'Zero-copy direct response handling error [streamId: $_streamId]',
        error: e,
        stackTrace: stackTrace,
      );
      if (!_responseController.isClosed) {
        _responseController.addError(e, stackTrace);
      }
    }
  }

  /// Processes response data (serialization mode only).
  void _processResponseData(List<int> messageBytes) {
    if (_isZeroCopy) {
      _logger?.logRpcWarning(
        message: 'Serialized response received in zero-copy mode, ignoring',
        methodPath: _methodPath,
        streamId: _streamId,
      );
      return;
    }

    _logger?.internal(
      'Received response payload: ${messageBytes.length} bytes [streamId: $_streamId]',
    );

    try {
      final uint8Message = messageBytes is Uint8List
          ? messageBytes
          : Uint8List.fromList(messageBytes);

      final messages = _parser!(uint8Message);
      _logger?.internal(
        'Parser extracted ${messages.length} messages from frame [streamId: $_streamId]',
      );

      for (var msgBytes in messages) {
        try {
          _logger?.internal(
            'Deserializing response of ${msgBytes.length} bytes [streamId: $_streamId]',
          );
          final response = _responseCodec!.deserialize(msgBytes);

          final rpcMessage = RpcMessage.withPayload<TResponse>(response);

          if (!_responseController.isClosed) {
            _responseController.add(rpcMessage);
            _logger?.internal(
              'Deserialized response added to stream [streamId: $_streamId]',
            );
          } else {
            _logger?.warning(
              'Cannot add response to closed controller [streamId: $_streamId]',
            );
          }
        } catch (e, stackTrace) {
          _logger?.error(
            'Failed to deserialize response [streamId: $_streamId]',
            error: e,
            stackTrace: stackTrace,
          );
          if (!_responseController.isClosed) {
            _responseController.addError(e, stackTrace);
          }
        }
      }
    } catch (e, stackTrace) {
      _logger?.error(
        'Failed to parse response [streamId: $_streamId]',
        error: e,
        stackTrace: stackTrace,
      );
      if (!_responseController.isClosed) {
        _responseController.addError(e, stackTrace);
      }
    }
  }

  /// Sends a request to the server.
  Future<void> send(TRequest request) async {
    if (!_isActive) {
      _logger?.warning('Attempted to send request on inactive processor');
      return;
    }

    if (!_requestController.isClosed) {
      _requestController.add(request);
    } else {
      _logger?.warning('Attempted to send request to closed controller');
    }
  }

  /// Finishes sending requests.
  Future<void> finishSending() async {
    if (!_isActive) return;

    _logger?.internal(
      'Finishing request send for $_methodPath [streamId: $_streamId]',
    );

    if (!_requestController.isClosed) {
      await _requestController.close();
    }
  }

  /// Closes the processor and releases resources.
  Future<void> close() async {
    if (!_isActive) return;

    _logger?.internal(
      'Closing CallProcessor for $_methodPath [streamId: $_streamId]',
    );
    _isActive = false;

    await _requestSubscription?.cancel();
    _requestSubscription = null;

    await _responseSubscription?.cancel();
    _responseSubscription = null;

    await _cancellationSubscription?.cancel();
    _cancellationSubscription = null;

    if (!_requestController.isClosed) {
      _requestController.close();
    }

    if (!_responseController.isClosed) {
      _responseController.close();
    }
  }
}

nogipx / rpc_dart / 23169302222

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous