setFile method

Future<Result> setFile(String procedure, WampFileSource source, { int chunkSize = defaultWampFileChunkSize, Duration timeout = const Duration(minutes: 5), CallOptions? options, })

Implementation

Future<Result> setFile(
  String procedure,
  WampFileSource source, {
  int chunkSize = defaultWampFileChunkSize,
  Duration timeout = const Duration(minutes: 5),
  CallOptions? options,
}) async {
  if (timeout.isNegative) {
    throw ArgumentError.value(timeout, 'timeout', 'must not be negative');
  }
  final metadata = WampFileMetadata(
    name: source.name,
    size: source.length,
    chunkSize: chunkSize,
    contentType: source.contentType,
    sha256Digest: source.sha256Digest,
    custom: source.custom,
  );
  final cancel = Completer<String>();
  final call = startProgressiveCall(
    procedure,
    argumentsKeywords: <String, dynamic>{
      wampFileMetadataKey: metadata.toJson(),
    },
    options: _fileCallOptions(options, timeout),
    cancelCompleter: cancel,
    enableFileSegments: source.nativePath != null,
  );
  Result? completedFinalResult;
  Object? finalError;
  StackTrace? finalStackTrace;
  final finalResultCompletion = call.results
      .firstWhere((result) => !result.isProgressive())
      .then<void>(
        (result) {
          completedFinalResult = result;
        },
        onError: (Object error, StackTrace stackTrace) {
          finalError = error;
          finalStackTrace = stackTrace;
        },
      );

  Future<Result> awaitFinalResult() async {
    await finalResultCompletion;
    final error = finalError;
    if (error != null) {
      return Future<Result>.error(error, finalStackTrace);
    }
    final result = completedFinalResult;
    if (result == null) {
      throw WampFileTransferException(
        'Remote procedure closed without a final result',
      );
    }
    return result;
  }

  Future<void> rejectEarlyFinalResult() async {
    final error = finalError;
    if (error != null) {
      await Future<void>.error(error, finalStackTrace);
    }
    if (completedFinalResult != null) {
      throw WampFileTransferException(
        'Remote procedure completed before the file source was exhausted',
      );
    }
  }

  var sentBytes = 0;
  Uint8List? pending;
  try {
    final nativePath = source.nativePath;
    if (nativePath != null &&
        source.length > 0 &&
        call.supportsFileSegments) {
      final nativeSource = call.openFileSource(nativePath, source.length);
      try {
        var offset = 0;
        while (offset < source.length) {
          final remaining = source.length - offset;
          final length = remaining < chunkSize ? remaining : chunkSize;
          if (offset + length == source.length) {
            call.finishFileSegment(
              nativeSource,
              offset: offset,
              length: length,
            );
          } else {
            call.sendFileSegment(
              nativeSource,
              offset: offset,
              length: length,
            );
            await call.drain();
            await rejectEarlyFinalResult();
          }
          offset += length;
        }
      } finally {
        nativeSource.close();
      }
      return await awaitFinalResult();
    }
    final chunks =
        source.openReadChunks?.call(chunkSize) ??
        _rechunkBytes(source.openRead(), chunkSize);
    await for (final chunk in chunks) {
      if (chunk.length > chunkSize) {
        throw WampFileTransferException(
          'Source emitted a ${chunk.length}-byte chunk above the '
          '$chunkSize-byte limit',
        );
      }
      sentBytes += chunk.length;
      if (sentBytes > source.length) {
        throw WampFileTransferException(
          'Source emitted more than its declared ${source.length} bytes',
        );
      }
      final previous = pending;
      if (previous != null) {
        call.sendLazyChunk(_fileChunkPayload(previous));
        await call.drain();
        await rejectEarlyFinalResult();
      }
      pending = chunk;
    }
    if (sentBytes != source.length) {
      throw WampFileTransferException(
        'Source emitted $sentBytes of its declared ${source.length} bytes',
      );
    }
    call.finishLazy(_fileChunkPayload(pending ?? Uint8List(0)));
    return await awaitFinalResult();
  } catch (error) {
    if (!cancel.isCompleted) {
      cancel.complete(CancelOptions.modeKillNoWait);
    }
    unawaited(finalResultCompletion);
    rethrow;
  }
}