1- // Copyright 2017 Google Inc. All Rights Reserved.
1+ // Copyright 2017 Google Inc. All Rights Reserved.
22//
33// Licensed under the Apache License, Version 2.0 (the "License");
44// you may not use this file except in compliance with the License.
1212// See the License for the specific language governing permissions and
1313// limitations under the License.
1414
15+ using Google . Apis . Http ;
1516using Google . Apis . Services ;
17+ using Google . Apis . Upload ;
1618using System ;
1719using System . IO ;
20+ using System . Linq ;
1821using System . Net . Http ;
22+ using System . Threading ;
23+ using System . Threading . Tasks ;
1924using static Google . Apis . Storage . v1 . ObjectsResource ;
20- using Google . Apis . Upload ;
2125
2226namespace Google . Cloud . Storage . V1
2327{
@@ -26,12 +30,157 @@ namespace Google.Cloud.Storage.V1
2630 /// </summary>
2731 internal sealed class CustomMediaUpload : InsertMediaUpload
2832 {
33+ private readonly Stream _stream ;
34+ private readonly Crc32cHashInterceptor _interceptor ;
35+ private readonly IClientService _service ;
36+ private readonly HashingStream _hashingStream ;
37+
2938 public CustomMediaUpload ( IClientService service , Apis . Storage . v1 . Data . Object body , string bucket ,
30- Stream stream , string contentType )
31- : base ( service , body , bucket , stream , contentType )
39+ Stream stream , string contentType , UploadObjectOptions options )
40+ : base ( service , body , bucket , new HashingStream ( stream ) , contentType )
3241 {
42+ _stream = stream ;
43+ _service = service ;
44+ var validationMode = options ? . UploadValidationMode ?? UploadObjectOptions . DefaultValidationMode ;
45+ _hashingStream = ( HashingStream ) ContentStream ;
46+ _interceptor = new Crc32cHashInterceptor ( this , _hashingStream , _service , validationMode ) ;
47+ _service ? . HttpClient ? . MessageHandler ? . AddExecuteInterceptor ( _interceptor ) ;
3348 }
3449
3550 internal new ResumableUploadOptions Options => base . Options ;
51+
52+ private sealed class Crc32cHashInterceptor : IHttpExecuteInterceptor
53+ {
54+ private const string GoogleHashHeader = "x-goog-hash" ;
55+ private const int ReadBufferSize = 81920 ;
56+ private readonly IClientService _service ;
57+ private readonly CustomMediaUpload _owner ;
58+ private Uri _uploadUri ;
59+ private readonly UploadValidationMode _validationMode ;
60+ private readonly HashingStream _hashingStream ;
61+
62+ public Crc32cHashInterceptor ( CustomMediaUpload owner , HashingStream hashingStream , IClientService service , UploadValidationMode validationMode )
63+ {
64+ _hashingStream = hashingStream ;
65+ _service = service ;
66+ _owner = owner ;
67+ _validationMode = validationMode ;
68+ _owner . UploadSessionData += OnSessionData ;
69+ _owner . ProgressChanged += OnProgressChanged ;
70+ }
71+
72+ public Task InterceptAsync ( HttpRequestMessage request , CancellationToken cancellationToken )
73+ {
74+ if ( _uploadUri != null && ! _uploadUri . Equals ( request . RequestUri ) )
75+ {
76+ return Task . CompletedTask ;
77+ }
78+
79+ if ( request . Method == System . Net . Http . HttpMethod . Put && request . Content ? . Headers . Contains ( "Content-Range" ) is true )
80+ {
81+ var rangeHeader = request . Content . Headers . GetValues ( "Content-Range" ) . First ( ) ;
82+
83+ if ( IsFinalChunk ( rangeHeader ) )
84+ {
85+ if ( _validationMode != UploadValidationMode . None )
86+ {
87+ var calculatedHash = _hashingStream . GetBase64Hash ( ) ;
88+ request . Headers . TryAddWithoutValidation ( GoogleHashHeader , $ "crc32c={ calculatedHash } ") ;
89+ }
90+ }
91+ }
92+ return Task . CompletedTask ;
93+ }
94+
95+ private void OnSessionData ( IUploadSessionData data )
96+ {
97+ _uploadUri = data . UploadUri ;
98+ _owner . UploadSessionData -= OnSessionData ;
99+ }
100+
101+ private void OnProgressChanged ( IUploadProgress progress )
102+ {
103+ if ( progress . Status == UploadStatus . Completed || progress . Status == UploadStatus . Failed )
104+ {
105+ // Clean up when upload is finished.
106+ _service ? . HttpClient ? . MessageHandler ? . RemoveExecuteInterceptor ( this ) ;
107+ _owner . ProgressChanged -= OnProgressChanged ;
108+ }
109+ }
110+
111+ private bool IsFinalChunk ( string rangeHeader )
112+ {
113+ // Expected format: "bytes {start}-{end}/{total}" or "bytes */{total}" for the final request.
114+ // We are interested in the final chunk of a known-size upload.
115+ const string prefix = "bytes " ;
116+ if ( ! rangeHeader . StartsWith ( prefix , StringComparison . OrdinalIgnoreCase ) )
117+ {
118+ return false ;
119+ }
120+
121+ ReadOnlySpan < char > span = rangeHeader . AsSpan ( prefix . Length ) ;
122+ int slashIndex = span . IndexOf ( '/' ) ;
123+ if ( slashIndex == - 1 )
124+ {
125+ return false ;
126+ }
127+
128+ var totalSpan = span . Slice ( slashIndex + 1 ) ;
129+ if ( totalSpan . IsEmpty || totalSpan [ 0 ] == '*' )
130+ {
131+ return false ;
132+ }
133+
134+ if ( ! long . TryParse ( totalSpan . ToString ( ) , System . Globalization . NumberStyles . Integer , System . Globalization . CultureInfo . InvariantCulture , out long totalSize ) )
135+ {
136+ return false ;
137+ }
138+
139+ var rangeSpan = span . Slice ( 0 , slashIndex ) ;
140+ int dashIndex = rangeSpan . IndexOf ( '-' ) ;
141+ if ( dashIndex == - 1 )
142+ {
143+ return false ;
144+ }
145+
146+ var endByteSpan = rangeSpan . Slice ( dashIndex + 1 ) ;
147+ if ( ! long . TryParse ( endByteSpan . ToString ( ) , System . Globalization . NumberStyles . Integer , System . Globalization . CultureInfo . InvariantCulture , out long endByte ) )
148+ {
149+ return false ;
150+ }
151+
152+ // If endByte is the last byte of the file, it's the final chunk.
153+ return ( endByte + 1 ) == totalSize ;
154+ }
155+ }
156+
157+ private sealed class HashingStream : Stream
158+ {
159+ private readonly Stream _inner ;
160+ private readonly Crc32c _hasher = new Crc32c ( ) ;
161+
162+ public HashingStream ( Stream inner ) => _inner = inner ;
163+
164+ public override int Read ( byte [ ] buffer , int offset , int count )
165+ {
166+ int bytesRead = _inner . Read ( buffer , offset , count ) ;
167+ if ( bytesRead > 0 )
168+ {
169+ _hasher . UpdateHash ( buffer , offset , bytesRead ) ;
170+ }
171+ return bytesRead ;
172+ }
173+
174+ public string GetBase64Hash ( ) => Convert . ToBase64String ( _hasher . GetHash ( ) ) ;
175+ public override bool CanRead => _inner . CanRead ;
176+ public override bool CanSeek => _inner . CanSeek ;
177+ public override bool CanWrite => _inner . CanWrite ;
178+ public override long Length => _inner . Length ;
179+ public override long Position { get => _inner . Position ; set => _inner . Position = value ; }
180+ public override void Flush ( ) => _inner . Flush ( ) ;
181+ public override long Seek ( long offset , SeekOrigin origin ) => _inner . Seek ( offset , origin ) ;
182+ public override void SetLength ( long value ) => _inner . SetLength ( value ) ;
183+ public override void Write ( byte [ ] buffer , int offset , int count ) => _inner . Write ( buffer , offset , count ) ;
184+ }
36185 }
37186}
0 commit comments