GraphServiceClient: LargeFileUploadTask with a Stream

Question

I'm using a library to encrypt files. The library takes an input Stream and an output Stream.

I'd like to use the library to encrypt data and upload them to OneDrive. I should use LargeFileUploadTask to upload the file to onedrive but I don't know if it is possible to avoid writing encrypted data to a temporary file before doing the actual upload.

Is there a way I can get a Stream to just write to while uploading the data to OneDrive?

var encrypter = new StreamEncrypter("password");
using (var sourceStream = await OpenStream(input)) {
    using (var destStream = await OpenStreamForWrite(output)) {
        await encrypter.Encrypt(sourceStream, destStream);
    }
}

Answer 1

Found a way, as suggested I implemented my own Stream. The size of the destination file must be known before the upload (in my scenario I can compute the final size because I'm using AES encryption). For now I didn't bother to implement the Seek method (probably it gets called only when the upload is resumed because I see it called just once with 0 as the requested position).

The test progam:

var appOptions = new PublicClientApplicationOptions()
{
    ClientName = appName,
    ClientId = appId,
    TenantId = tenantId,
    RedirectUri = redirectUri
};
var app = PublicClientApplicationBuilder.CreateWithApplicationOptions(appOptions).Build();

var storageProperties = new StorageCreationPropertiesBuilder("UserTokenCache.txt", "./cache")
    .WithLinuxKeyring(
        "com.contoso.devtools.tokencache",
        MsalCacheHelper.LinuxKeyRingDefaultCollection,
        "MSAL token cache for all Contoso dev tool apps.",
        new KeyValuePair<string, string>("Version", "1"),
        new KeyValuePair<string, string>("ProductGroup", "MyApps"))
    .WithMacKeyChain(
        "myapp_msal_service",
        "myapp_msal_account")
    .Build();
var cacheHelper = await MsalCacheHelper.CreateAsync(storageProperties);
cacheHelper.VerifyPersistence();
cacheHelper.RegisterCache(app.UserTokenCache);

var accounts = await app.GetAccountsAsync();
AuthenticationResult result;
try {
    result = await app.AcquireTokenSilent(scopes, accounts.FirstOrDefault())
                .ExecuteAsync();
} catch (MsalUiRequiredException) {
    result = await app.AcquireTokenInteractive(scopes)
                .ExecuteAsync();
}

var graphClient = new GraphServiceClient(new AuthenticationProvider(result.AccessToken));

using (var fileStream = System.IO.File.OpenRead(@"c:\test\test.zip")) {
    UploadSession uploadSession = await graphClient.Me.Drive.Root.ItemWithPath("/test/test.zip").CreateUploadSession().Request().PostAsync();
    using (var oneDriveStream = new OneDriveStream(uploadSession, fileStream.Length)) {
        byte[] buffer = new byte[4096];
        int read;
        while ((read = await fileStream.ReadAsync(buffer, 0, buffer.Length)) > 0) {
            await oneDriveStream.WriteAsync(buffer, 0, read);
        }
        await oneDriveStream.WaitTask();
    }
}

The OneDriveStream class:

    internal class OneDriveStream : Stream
    {
        private readonly Task? _writeTask;
        private long _pos;

        private readonly long _cacheSize = 1310720;
        private readonly List<byte> _buffer = new();
        private readonly SemaphoreSlim _bufferSema = new(1,1);

        public WriterStream(UploadSession session, long length)
        {
            Length = length;
            _writeTask = new LargeFileUploadTask<DriveItem>(session, this, (int)_cacheSize).UploadAsync();
        }

        public override bool CanRead => true;

        public override bool CanSeek => true;

        public override bool CanWrite => true;

        public override long Length { get; }

        public override long Position {
            get => _pos;
            set => _pos = value;
        }

        protected override void Dispose(bool disposing)
        {
            _writeTask?.Dispose();
        }

        public override void Flush()
        {
        }

        public override int Read(byte[] buffer, int offset, int count)
        {
            if (_pos >= Length)
                return 0;

            _bufferSema.Wait();
            int readCount = 0;
            for (int i = 0; i < count; i++) {
                if (_buffer.Count > 0) {
                    buffer[offset + i] = _buffer[0];
                    _buffer.RemoveAt(0);

                    _pos++;
                    readCount++;
                    if (_pos >= Length)
                        break;
                } else {
                    _bufferSema.Release();
                    Thread.Sleep(20);
                    _bufferSema.Wait();
                    i--;
                }
            }
            _bufferSema.Release();
            return readCount;
        }

        public override long Seek(long offset, SeekOrigin origin)
        {
            return offset;
        }

        public override void SetLength(long value)
        {
            throw new NotImplementedException();
        }

        public override void Write(byte[] buffer, int offset, int count)
        {
            while(_buffer.Count > 0)
                Thread.Sleep(10);

            _bufferSema.Wait();
            for (int i = 0; i < count; i++) {
                _buffer.Add(buffer[offset + i]);
            }
            _bufferSema.Release();
        }

        public async Task WaitTask()
        {
            if (_writeTask != null)
                await _writeTask;
        }
    }