consumes method

@override Stream<Uint8List> consumes(Context ctx, String urn, Execution exec, Uint8List inbound,)

Consumes sends the request and returns a stream of raw response bytes. The stream is closed when the server finishes streaming. urn Actual uniform resource Domain Name. execution Service reference. inbound Input arguments. Output stream of payload chunks

Implementation

@override
Stream<Uint8List> consumes(
    Context ctx, String urn, Execution exec, Uint8List inbound) async* {
  final uri = Uri.parse(ctx.getAttribute(addressKey));
  final request = Request('POST', Uri.https(uri.authority, '/mpc/invoke'));
  request.bodyBytes = inbound;
  request.headers.addAll(ctx.getAttribute(headerKey));
  request.headers.addAll(ctx.attachments());
  request.headers['mesh-urn'] = urn;
  request.headers['Accept'] = 'text/event-stream';
  final response = await hc.send(request);
  if (response.statusCode != 200) {
    throw Status.system.err(r: '${response.statusCode}');
  }
  final lines =
      response.stream.transform(utf8.decoder).transform(const LineSplitter());
  await for (final line in lines) {
    if (line.startsWith('data:')) {
      final data = line.substring(5).trim();
      if (data.isNotEmpty) {
        yield utf8.encode(data);
      }
    }
  }
}