Hi3Helper.Http/IO.cs at main · CollapseLauncher/Hi3Helper.Http · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
using System;
using System.Buffers;
using System.IO;
using System.Threading;
using System.Threading.Tasks;

// ReSharper disable InconsistentNaming
// ReSharper disable AccessToModifiedClosure

namespace Hi3Helper.Http
{
    internal static class IO
    {
        internal const int StreamRWBufferSize = 64 << 10;

        internal static async Task WriteStreamToFileChunkSessionAsync(
            ChunkSession              session,
            DownloadSpeedLimiter?     downloadSpeedLimiter,
            int                       threadSize,
            HttpResponseInputStream?  networkStream,
            bool                      isNetworkStreamFromExternal,
            Stream                    fileStream,
            DownloadProgress          downloadProgress,
            DownloadProgressDelegate? progressDelegateAsync,
            CancellationToken         token)
        {
            int currentRetry = 0;

        StartWrite:
            byte[] buffer = ArrayPool<byte>.Shared.Rent(16 << 10);
            CancellationTokenSource? timeoutToken = null;
            CancellationTokenSource? coopToken = null;

            try
            {
                if (session.CurrentPositions.End != 0 && session.CurrentPositions.Start >= session.CurrentPositions.End)
                {
                    return;
                }

                if (!isNetworkStreamFromExternal || (isNetworkStreamFromExternal && currentRetry > 0))
                {
                    networkStream = await CreateStreamFromSessionAsync(session, token);
                }

                if (networkStream == null)
                {
                    throw new NullReferenceException("networkStream argument returns null!");
                }

                if (fileStream.CanSeek && session.CurrentPositions.End + 1 > fileStream.Length)
                {
                    fileStream.SetLength(session.CurrentPositions.Start);
                }

                if (fileStream.CanSeek)
                {
                    fileStream.Seek(session.CurrentPositions.Start, SeekOrigin.Begin);
                }

                timeoutToken = new CancellationTokenSource(session.TimeoutAfterInterval);
                coopToken = CancellationTokenSource.CreateLinkedTokenSource(timeoutToken.Token, token);

                int read;
                while ((read = await networkStream.ReadAsync(buffer, coopToken.Token)) > 0)
                {
                    await (downloadSpeedLimiter?.AddBytesOrWaitAsync(read, token) ?? ValueTask.CompletedTask);
                    await fileStream.WriteAsync(buffer, 0, read, coopToken.Token);

                    session.CurrentPositions.AdvanceStartOffset(read);
                    session.CurrentMetadata?.UpdateLastEndOffset(session.CurrentPositions);
                    downloadProgress.AdvanceBytesDownloaded(read);
                    progressDelegateAsync?.Invoke(read, downloadProgress);

                    timeoutToken.Dispose();
                    coopToken.Dispose();

                    timeoutToken = new CancellationTokenSource(session.TimeoutAfterInterval);
                    coopToken = CancellationTokenSource.CreateLinkedTokenSource(timeoutToken.Token, token);

                    currentRetry = 0;
                }
            }
            catch (TaskCanceledException) when (token.IsCancellationRequested)
            {
                throw;
            }
            catch (OperationCanceledException) when (token.IsCancellationRequested)
            {
                throw;
            }
            catch (NullReferenceException)
            {
                throw;
            }
            catch (Exception)
            {
                currentRetry++;
                if (currentRetry > session.RetryMaxAttempt)
                {
                    throw;
                }

                await Task.Delay(session.RetryAttemptInterval, token);
                goto StartWrite;
            }
            finally
            {
                if (networkStream != null)
                {
                    await networkStream.DisposeAsync();
                }

                ArrayPool<byte>.Shared.Return(buffer);

                timeoutToken?.Dispose();
                coopToken?.Dispose();
            }
        }

        private static async ValueTask<HttpResponseInputStream?> CreateStreamFromSessionAsync(ChunkSession session,
            CancellationToken token)
        {
            // Assign the url and throw if null
            Uri? fileUri = session.CurrentMetadata?.Url;
            if (fileUri == null)
            {
                throw new NullReferenceException("Metadata was found to be null and it shouldn't happen!");
            }

            // Create the network stream instance
            HttpResponseInputStream? stream = await HttpResponseInputStream
                .CreateStreamAsync(
                    session.CurrentHttpClient,
                    fileUri,
                    session.CurrentPositions.Start,
                    session.CurrentPositions.End,
                    session.TimeoutAfterInterval,
                    session.RetryAttemptInterval,
                    session.RetryMaxAttempt,
                    token);

            // Return the stream
            return stream;
        }
    }
}