stream method

Future<MessagingStream> stream({ required Participant to, required String type, required Map<String, dynamic> message, Uint8List? attachment, })

Implementation

Future<MessagingStream> stream({
  required Participant to,
  required String type,
  required Map<String, dynamic> message,
  Uint8List? attachment,
}) async {
  final input = _MessagingStreamInput();
  unawaited(
    input.add(
      JsonContent(
        json: <String, dynamic>{
          'to_participant_id': to.id,
          'type': type,
          'message_json': jsonEncode(message),
          if (attachment != null) 'attachment_base64': base64Encode(attachment),
        },
      ),
    ),
  );
  final output = await room.invoke(toolkit: 'messaging', tool: 'stream', input: ToolStreamInput(input.stream()));
  if (output is! ToolStreamOutput) {
    input.close();
    throw RoomServerException('unexpected return type from messaging.stream');
  }
  final inputClosed = output.inputClosed;
  final iterator = StreamIterator<Content>(output.stream);
  if (!await iterator.moveNext()) {
    input.close();
    await iterator.cancel();
    throw RoomServerException('messaging.stream closed before acceptance');
  }
  final accepted = iterator.current;
  if (accepted is! JsonContent || accepted.json['kind'] != 'accepted' || accepted.json['stream_id'] is! String) {
    input.close();
    await iterator.cancel();
    throw RoomServerException('messaging.stream did not acknowledge acceptance');
  }
  late MessagingStream result;
  result = MessagingStream._(
    streamId: accepted.json['stream_id'] as String,
    remoteParticipantId: to.id,
    input: input,
    output: iterator,
    inputClosed: inputClosed,
    onClosed: () => _streams.remove(result),
  );
  _streams.add(result);
  return result;
}