feat: Allow a new upload session to be initiated as a single method call

This allows the content length to be (optionally) specified, which is then
propagated via X-Upload-Content-Length.

This is currently just a prototype for comment - I'd want to add
integration tests, some unit tests, and obviously XML docs.

apis/Google.Cloud.Storage.V1/Google.Cloud.Storage.V1/StorageClient.UploadObject.cs

@@ -153,5 +153,38 @@ public virtual Task<Object> UploadObjectAsync(
         {
             throw new NotImplementedException();
         }
+
+        /// <summary>
+        /// 
+        /// </summary>
+        /// <param name="destination"></param>
+        /// <param name="contentLength"></param>
+        /// <param name="cancellationToken">The token to monitor for cancellation requests.</param>
+        /// <returns></returns>
+        /// <exception cref="NotImplementedException"></exception>
+        public virtual Task<Uri> InitiateUploadSessionAsync(
+            Object destination,
+            long? contentLength,
+            CancellationToken cancellationToken = default) =>
+            throw new NotImplementedException();
+
+        /// <summary>
+        /// 
+        /// </summary>
+        /// <param name="bucket">The name of the bucket containing the object. Must not be null.</param>
+        /// <param name="objectName">The name of the object within the bucket. Must not be null.</param>
+        /// <param name="contentType">The content type of the object. This should be a MIME type
+        /// such as "text/html" or "application/octet-stream". May be null.</param>
+        /// <param name="contentLength"></param>
+        /// <param name="cancellationToken">The token to monitor for cancellation requests.</param>
+        /// <returns></returns>
+        /// <exception cref="NotImplementedException"></exception>
+        public virtual Task<Uri> InitiateUploadSessionAsync(
+            string bucket,
+            string objectName,
+            string contentType,
+            long? contentLength,
+            CancellationToken cancellationToken = default) =>
+            throw new NotImplementedException();
     }
 }

apis/Google.Cloud.Storage.V1/Google.Cloud.Storage.V1/StorageClientImpl.UploadObject.cs

@@ -17,8 +17,10 @@
 using Google.Apis.Upload;
 using System;
 using System.IO;
+using System.Runtime.InteropServices.WindowsRuntime;
 using System.Threading;
 using System.Threading.Tasks;
+using static Google.Apis.Storage.v1.ObjectsResource;
 using Object = Google.Apis.Storage.v1.Data.Object;
 
 namespace Google.Cloud.Storage.V1
@@ -103,6 +105,29 @@ public override Task<Object> UploadObjectAsync(
             IProgress<IUploadProgress> progress = null) =>
             new UploadHelper(this, destination, source, options, progress).ExecuteAsync(cancellationToken);
 
+        /// <inheritdoc />
+        public override Task<Uri> InitiateUploadSessionAsync(Object destination, long? contentLength, CancellationToken cancellationToken = default)
+        {
+            // We could potentially do a single validation, but the reasons for preventing negative and zero
+            // values are somewhat different.
+            GaxPreconditions.CheckNonNegative(contentLength, nameof(contentLength));
+            if (contentLength == 0)
+            {
+                throw new ArgumentOutOfRangeException("A content length of 0 cannot be enforced. Use a null content length for 'any length'.");
+            }
+            ValidateObject(destination, nameof(destination));
+            var upload = CreateObjectUploader(destination, new LengthOnlyStream(contentLength));
+            return upload.InitiateSessionAsync(cancellationToken);
+        }
+
+        /// <inheritdoc />
+        public override Task<Uri> InitiateUploadSessionAsync(string bucket, string objectName, string contentType, long? contentLength, CancellationToken cancellationToken = default)
+        {
+            ValidateBucketName(bucket);
+            var obj = new Object { Bucket = bucket, Name = objectName, ContentType = contentType };
+            return InitiateUploadSessionAsync(obj, contentLength, cancellationToken);
+        }
+
         /// <summary>
         /// Helper class to provide common context between sync and async operations. Helps avoid quite so much duplicate code...
         /// </summary>
@@ -192,5 +217,33 @@ internal async Task<Object> ExecuteAsync(CancellationToken cancellationToken)
                 return result;
             }
         }
+
+        private sealed class LengthOnlyStream : Stream
+        {
+            private readonly long? _length;
+            internal LengthOnlyStream(long? length) => _length = length;
+
+            public override long Length => _length ?? throw new NotSupportedException();
+            public override bool CanSeek => _length.HasValue;
+
+            public override bool CanRead => throw new NotImplementedException();
+            public override bool CanWrite => throw new NotImplementedException();
+
+            public override long Position
+            {
+                get => throw new NotImplementedException();
+                set => throw new NotImplementedException();
+            }
+
+            public override void Flush() => throw new NotImplementedException();
+            public override int Read(byte[] buffer, int offset, int count) =>
+                throw new NotImplementedException();
+            public override long Seek(long offset, SeekOrigin origin) =>
+                throw new NotImplementedException();
+            public override void SetLength(long value) =>
+                throw new NotImplementedException();
+            public override void Write(byte[] buffer, int offset, int count) =>
+                throw new NotImplementedException();
+        }
     }
 }