consumeData method

void consumeData({ required String id, required String dataProducerId, required SctpStreamParameters sctpStreamParameters, String label = '', String protocol = '', Map<String, dynamic> appData = const <String, dynamic>{}, })

Implementation

void consumeData({
  required String id,
  required String dataProducerId,
  required SctpStreamParameters sctpStreamParameters,
  String label = '',
  String protocol = '',
  Map<String, dynamic> appData = const <String, dynamic>{},
}) {
  _logger.debug('consumeData()');

  sctpStreamParameters = SctpStreamParameters.copy(sctpStreamParameters);

  if (_closed) {
    throw ('closed');
  } else if (_direction != Direction.recv) {
    throw ('not a receiving Transport');
  } else if (_maxSctpMessageSize == null) {
    throw ('SCTP not enabled by remote Transport');
  } else if (listeners('connect').isEmpty && _connectionState == 'new') {
    throw ('no "connect" listener set into this transport');
  }

  // This may throw.
  Ortc.validateSctpStreamParameters(sctpStreamParameters);

  // Enqueue command.

  _flexQueue.addTask(FlexTaskAdd(
    id: id,
    message: 'transport.consumeData()',
    execFun: () async {
      HandlerReceiveDataChannelResult receiveDataChannelResult =
          await _handler.receiveDataChannel(HandlerReceiveDataChannelOptions(
        sctpStreamParameters: sctpStreamParameters,
        label: label,
        protocol: protocol,
      ));

      DataConsumer dataConsumer = DataConsumer(
          id: id,
          dataProducerId: dataProducerId,
          dataChannel: receiveDataChannelResult.dataChannel,
          sctpStreamParameters: sctpStreamParameters,
          appData: appData);

      _dataConsumers[dataConsumer.id] = dataConsumer;
      _handleDataConsumer(dataConsumer);

      // Emit observer event.
      _observer.safeEmit('newdataconsumer', {
        'dataConsumer': dataConsumer,
      });

      dataConsumerCallback?.call(dataConsumer);
    },
  ));
}