Fix customer serializer being used for internal types

Microsoft.Azure.Cosmos/src/Batch/BatchAsyncBatcher.cs

@@ -24,7 +24,7 @@ namespace Microsoft.Azure.Cosmos
     /// <seealso cref="ItemBatchOperation"/>
     internal class BatchAsyncBatcher
     {
-        private readonly CosmosSerializer cosmosSerializer;
+        private readonly CosmosSerializerCore serializerCore;
         private readonly List<ItemBatchOperation> batchOperations;
         private readonly BatchAsyncBatcherExecuteDelegate executor;
         private readonly BatchAsyncBatcherRetryDelegate retrier;
@@ -39,7 +39,7 @@ internal class BatchAsyncBatcher
         public BatchAsyncBatcher(
             int maxBatchOperationCount,
             int maxBatchByteSize,
-            CosmosSerializer cosmosSerializer,
+            CosmosSerializerCore serializerCore,
             BatchAsyncBatcherExecuteDelegate executor,
             BatchAsyncBatcherRetryDelegate retrier)
         {
@@ -63,17 +63,17 @@ public BatchAsyncBatcher(
                 throw new ArgumentNullException(nameof(retrier));
             }
 
-            if (cosmosSerializer == null)
+            if (serializerCore == null)
             {
-                throw new ArgumentNullException(nameof(cosmosSerializer));
+                throw new ArgumentNullException(nameof(serializerCore));
             }
 
             this.batchOperations = new List<ItemBatchOperation>(maxBatchOperationCount);
             this.executor = executor;
             this.retrier = retrier;
             this.maxBatchByteSize = maxBatchByteSize;
             this.maxBatchOperationCount = maxBatchOperationCount;
-            this.cosmosSerializer = cosmosSerializer;
+            this.serializerCore = serializerCore;
         }
 
         public virtual bool TryAdd(ItemBatchOperation operation)
@@ -152,7 +152,7 @@ public virtual bool TryAdd(ItemBatchOperation operation)
                 try
                 {
                     PartitionKeyRangeBatchExecutionResult result = await this.executor(serverRequest, cancellationToken);
-                    using (PartitionKeyRangeBatchResponse batchResponse = new PartitionKeyRangeBatchResponse(serverRequest.Operations.Count, result.ServerResponse, this.cosmosSerializer))
+                    using (PartitionKeyRangeBatchResponse batchResponse = new PartitionKeyRangeBatchResponse(serverRequest.Operations.Count, result.ServerResponse, this.serializerCore))
                     {
                         foreach (ItemBatchOperation itemBatchOperation in batchResponse.Operations)
                         {
@@ -207,7 +207,7 @@ internal virtual async Task<Tuple<PartitionKeyRangeServerBatchRequest, ArraySegm
                   this.maxBatchByteSize,
                   this.maxBatchOperationCount,
                   ensureContinuousOperationIndexes: false,
-                  serializer: this.cosmosSerializer,
+                  serializerCore: this.serializerCore,
                   cancellationToken: cancellationToken).ConfigureAwait(false);
         }
     }

Microsoft.Azure.Cosmos/src/Batch/BatchAsyncContainerExecutor.cs

@@ -134,7 +134,7 @@ internal virtual async Task ValidateOperationAsync(
                 Debug.Assert(BatchAsyncContainerExecutor.ValidateOperationEPK(operation, itemRequestOptions));
             }
 
-            await operation.MaterializeResourceAsync(this.cosmosClientContext.CosmosSerializer, cancellationToken);
+            await operation.MaterializeResourceAsync(this.cosmosClientContext.SerializerCore, cancellationToken);
         }
 
         private static IDocumentClientRetryPolicy GetRetryPolicy(RetryOptions retryOptions)
@@ -236,7 +236,7 @@ private async Task<PartitionKeyRangeBatchExecutionResult> ExecuteAsync(
                         requestEnricher: requestMessage => BatchAsyncContainerExecutor.AddHeadersToRequestMessage(requestMessage, serverRequest.PartitionKeyRangeId),
                         cancellationToken: cancellationToken).ConfigureAwait(false);
 
-                    TransactionalBatchResponse serverResponse = await TransactionalBatchResponse.FromResponseMessageAsync(responseMessage, serverRequest, this.cosmosClientContext.CosmosSerializer).ConfigureAwait(false);
+                    TransactionalBatchResponse serverResponse = await TransactionalBatchResponse.FromResponseMessageAsync(responseMessage, serverRequest, this.cosmosClientContext.SerializerCore).ConfigureAwait(false);
 
                     return new PartitionKeyRangeBatchExecutionResult(serverRequest.PartitionKeyRangeId, serverRequest.Operations, serverResponse);
                 }
@@ -250,7 +250,7 @@ private BatchAsyncStreamer GetOrAddStreamerForPartitionKeyRange(string partition
                 return streamer;
             }
 
-            BatchAsyncStreamer newStreamer = new BatchAsyncStreamer(this.maxServerRequestOperationCount, this.maxServerRequestBodyLength, this.dispatchTimerInSeconds, this.timerPool, this.cosmosClientContext.CosmosSerializer, this.ExecuteAsync, this.ReBatchAsync);
+            BatchAsyncStreamer newStreamer = new BatchAsyncStreamer(this.maxServerRequestOperationCount, this.maxServerRequestBodyLength, this.dispatchTimerInSeconds, this.timerPool, this.cosmosClientContext.SerializerCore, this.ExecuteAsync, this.ReBatchAsync);
             if (!this.streamersByPartitionKeyRange.TryAdd(partitionKeyRangeId, newStreamer))
             {
                 newStreamer.Dispose();

Microsoft.Azure.Cosmos/src/Batch/BatchAsyncStreamer.cs

@@ -26,7 +26,7 @@ internal class BatchAsyncStreamer : IDisposable
         private readonly BatchAsyncBatcherExecuteDelegate executor;
         private readonly BatchAsyncBatcherRetryDelegate retrier;
         private readonly int dispatchTimerInSeconds;
-        private readonly CosmosSerializer cosmosSerializer;
+        private readonly CosmosSerializerCore serializerCore;
         private readonly CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();
         private volatile BatchAsyncBatcher currentBatcher;
         private TimerPool timerPool;
@@ -38,7 +38,7 @@ public BatchAsyncStreamer(
             int maxBatchByteSize,
             int dispatchTimerInSeconds,
             TimerPool timerPool,
-            CosmosSerializer cosmosSerializer,
+            CosmosSerializerCore serializerCore,
             BatchAsyncBatcherExecuteDelegate executor,
             BatchAsyncBatcherRetryDelegate retrier)
         {
@@ -67,9 +67,9 @@ public BatchAsyncStreamer(
                 throw new ArgumentNullException(nameof(retrier));
             }
 
-            if (cosmosSerializer == null)
+            if (serializerCore == null)
             {
-                throw new ArgumentNullException(nameof(cosmosSerializer));
+                throw new ArgumentNullException(nameof(serializerCore));
             }
 
             this.maxBatchOperationCount = maxBatchOperationCount;
@@ -78,7 +78,7 @@ public BatchAsyncStreamer(
             this.retrier = retrier;
             this.dispatchTimerInSeconds = dispatchTimerInSeconds;
             this.timerPool = timerPool;
-            this.cosmosSerializer = cosmosSerializer;
+            this.serializerCore = serializerCore;
             this.currentBatcher = this.CreateBatchAsyncBatcher();
 
             this.ResetTimer();
@@ -157,7 +157,7 @@ private BatchAsyncBatcher GetBatchToDispatchAndCreate()
 
         private BatchAsyncBatcher CreateBatchAsyncBatcher()
         {
-            return new BatchAsyncBatcher(this.maxBatchOperationCount, this.maxBatchByteSize, this.cosmosSerializer, this.executor, this.retrier);
+            return new BatchAsyncBatcher(this.maxBatchOperationCount, this.maxBatchByteSize, this.serializerCore, this.executor, this.retrier);
         }
     }
 }

Microsoft.Azure.Cosmos/src/Batch/BatchExecutor.cs

@@ -51,7 +51,7 @@ public async Task<TransactionalBatchResponse> ExecuteAsync(CancellationToken can
             SinglePartitionKeyServerBatchRequest serverRequest = await SinglePartitionKeyServerBatchRequest.CreateAsync(
                       serverRequestPartitionKey,
                       new ArraySegment<ItemBatchOperation>(this.inputOperations.ToArray()),
-                      this.clientContext.CosmosSerializer,
+                      this.clientContext.SerializerCore,
                       cancellationToken);
 
             return await this.ExecuteServerRequestAsync(serverRequest, cancellationToken);
@@ -89,7 +89,7 @@ private async Task<TransactionalBatchResponse> ExecuteServerRequestAsync(
                 return await TransactionalBatchResponse.FromResponseMessageAsync(
                     responseMessage,
                     serverRequest,
-                    this.clientContext.CosmosSerializer);
+                    this.clientContext.SerializerCore);
             }
         }
     }

Microsoft.Azure.Cosmos/src/Batch/ItemBatchOperation.cs

@@ -287,9 +287,9 @@ internal int GetApproximateSerializedLength()
         /// <summary>
         /// Materializes the operation's resource into a Memory{byte} wrapping a byte array.
         /// </summary>
-        /// <param name="serializer">Serializer to serialize user provided objects to JSON.</param>
+        /// <param name="serializerCore">Serializer to serialize user provided objects to JSON.</param>
         /// <param name="cancellationToken"><see cref="CancellationToken"/> for cancellation.</param>
-        internal virtual async Task MaterializeResourceAsync(CosmosSerializer serializer, CancellationToken cancellationToken)
+        internal virtual async Task MaterializeResourceAsync(CosmosSerializerCore serializerCore, CancellationToken cancellationToken)
         {
             if (this.body.IsEmpty && this.ResourceStream != null)
             {
@@ -361,14 +361,14 @@ public ItemBatchOperation(
         /// <summary>
         /// Materializes the operation's resource into a Memory{byte} wrapping a byte array.
         /// </summary>
-        /// <param name="serializer">Serializer to serialize user provided objects to JSON.</param>
+        /// <param name="serializerCore">Serializer to serialize user provided objects to JSON.</param>
         /// <param name="cancellationToken"><see cref="CancellationToken"/> for cancellation.</param>
-        internal override Task MaterializeResourceAsync(CosmosSerializer serializer, CancellationToken cancellationToken)
+        internal override Task MaterializeResourceAsync(CosmosSerializerCore serializerCore, CancellationToken cancellationToken)
         {
             if (this.body.IsEmpty && this.Resource != null)
             {
-                this.ResourceStream = serializer.ToStream(this.Resource);
-                return base.MaterializeResourceAsync(serializer, cancellationToken);
+                this.ResourceStream = serializerCore.ToStream(this.Resource);
+                return base.MaterializeResourceAsync(serializerCore, cancellationToken);
             }
 
             return Task.FromResult(true);

Microsoft.Azure.Cosmos/src/Batch/PartitionKeyRangeBatchResponse.cs

@@ -25,11 +25,11 @@ internal class PartitionKeyRangeBatchResponse : TransactionalBatchResponse
         /// </summary>
         /// <param name="originalOperationsCount">Original operations that generated the server responses.</param>
         /// <param name="serverResponse">Response from the server.</param>
-        /// <param name="serializer">Serializer to deserialize response resource body streams.</param>
+        /// <param name="serializerCore">Serializer to deserialize response resource body streams.</param>
         internal PartitionKeyRangeBatchResponse(
             int originalOperationsCount,
             TransactionalBatchResponse serverResponse,
-            CosmosSerializer serializer)
+            CosmosSerializerCore serializerCore)
         {
             this.StatusCode = serverResponse.StatusCode;
 
@@ -59,7 +59,7 @@ internal PartitionKeyRangeBatchResponse(
 
             this.ErrorMessage = errorMessageBuilder.Length > 2 ? errorMessageBuilder.ToString(0, errorMessageBuilder.Length - 2) : null;
             this.Operations = itemBatchOperations;
-            this.Serializer = serializer;
+            this.SerializerCore = serializerCore;
         }
 
         /// <summary>
@@ -70,7 +70,7 @@ internal PartitionKeyRangeBatchResponse(
         /// <inheritdoc />
         public override CosmosDiagnostics Diagnostics => this.serverResponse.Diagnostics;
 
-        internal override CosmosSerializer Serializer { get; }
+        internal override CosmosSerializerCore SerializerCore { get; }
 
         /// <summary>
         /// Gets the number of operation results.
@@ -98,7 +98,7 @@ public override TransactionalBatchOperationResult<T> GetOperationResultAtIndex<T
             T resource = default(T);
             if (result.ResourceStream != null)
             {
-                resource = this.Serializer.FromStream<T>(result.ResourceStream);
+                resource = this.SerializerCore.FromStream<T>(result.ResourceStream);
             }
 
             return new TransactionalBatchOperationResult<T>(result, resource);

Microsoft.Azure.Cosmos/src/Batch/PartitionKeyRangeServerBatchRequest.cs

@@ -18,13 +18,13 @@ internal sealed class PartitionKeyRangeServerBatchRequest : ServerBatchRequest
         /// <param name="partitionKeyRangeId">The partition key range id associated with all requests.</param>
         /// <param name="maxBodyLength">Maximum length allowed for the request body.</param>
         /// <param name="maxOperationCount">Maximum number of operations allowed in the request.</param>
-        /// <param name="serializer">Serializer to serialize user provided objects to JSON.</param>
+        /// <param name="serializerCore">Serializer to serialize user provided objects to JSON.</param>
         public PartitionKeyRangeServerBatchRequest(
             string partitionKeyRangeId,
             int maxBodyLength,
             int maxOperationCount,
-            CosmosSerializer serializer)
-            : base(maxBodyLength, maxOperationCount, serializer)
+            CosmosSerializerCore serializerCore)
+            : base(maxBodyLength, maxOperationCount, serializerCore)
         {
             this.PartitionKeyRangeId = partitionKeyRangeId;
         }
@@ -44,7 +44,7 @@ public PartitionKeyRangeServerBatchRequest(
         /// <param name="maxBodyLength">Desired maximum length of the request body.</param>
         /// <param name="maxOperationCount">Maximum number of operations allowed in the request.</param>
         /// <param name="ensureContinuousOperationIndexes">Whether to stop adding operations to the request once there is non-continuity in the operation indexes.</param>
-        /// <param name="serializer">Serializer to serialize user provided objects to JSON.</param>
+        /// <param name="serializerCore">Serializer to serialize user provided objects to JSON.</param>
         /// <param name="cancellationToken"><see cref="CancellationToken"/> representing request cancellation.</param>
         /// <returns>A newly created instance of <see cref="PartitionKeyRangeServerBatchRequest"/>.</returns>
         public static async Task<Tuple<PartitionKeyRangeServerBatchRequest, ArraySegment<ItemBatchOperation>>> CreateAsync(
@@ -53,10 +53,10 @@ public static async Task<Tuple<PartitionKeyRangeServerBatchRequest, ArraySegment
             int maxBodyLength,
             int maxOperationCount,
             bool ensureContinuousOperationIndexes,
-            CosmosSerializer serializer,
+            CosmosSerializerCore serializerCore,
             CancellationToken cancellationToken)
         {
-            PartitionKeyRangeServerBatchRequest request = new PartitionKeyRangeServerBatchRequest(partitionKeyRangeId, maxBodyLength, maxOperationCount, serializer);
+            PartitionKeyRangeServerBatchRequest request = new PartitionKeyRangeServerBatchRequest(partitionKeyRangeId, maxBodyLength, maxOperationCount, serializerCore);
             ArraySegment<ItemBatchOperation> pendingOperations = await request.CreateBodyStreamAsync(operations, cancellationToken, ensureContinuousOperationIndexes);
             return new Tuple<PartitionKeyRangeServerBatchRequest, ArraySegment<ItemBatchOperation>>(request, pendingOperations);
         }

Microsoft.Azure.Cosmos/src/Batch/ServerBatchRequest.cs

@@ -21,7 +21,7 @@ internal abstract class ServerBatchRequest
 
         private readonly int maxOperationCount;
 
-        private readonly CosmosSerializer serializer;
+        private readonly CosmosSerializerCore serializerCore;
 
         private ArraySegment<ItemBatchOperation> operations;
 
@@ -40,12 +40,12 @@ internal abstract class ServerBatchRequest
         /// </summary>
         /// <param name="maxBodyLength">Maximum length allowed for the request body.</param>
         /// <param name="maxOperationCount">Maximum number of operations allowed in the request.</param>
-        /// <param name="serializer">Serializer to serialize user provided objects to JSON.</param>
-        protected ServerBatchRequest(int maxBodyLength, int maxOperationCount, CosmosSerializer serializer)
+        /// <param name="serializerCore">Serializer to serialize user provided objects to JSON.</param>
+        protected ServerBatchRequest(int maxBodyLength, int maxOperationCount, CosmosSerializerCore serializerCore)
         {
             this.maxBodyLength = maxBodyLength;
             this.maxOperationCount = maxOperationCount;
-            this.serializer = serializer;
+            this.serializerCore = serializerCore;
         }
 
         public IReadOnlyList<ItemBatchOperation> Operations => this.operations;
@@ -87,7 +87,7 @@ protected async Task<ArraySegment<ItemBatchOperation>> CreateBodyStreamAsync(
                     break;
                 }
 
-                await operation.MaterializeResourceAsync(this.serializer, cancellationToken);
+                await operation.MaterializeResourceAsync(this.serializerCore, cancellationToken);
                 materializedCount++;
 
                 previousOperationIndex = operation.OperationIndex;

Microsoft.Azure.Cosmos/src/Batch/SinglePartitionKeyServerBatchRequest.cs

@@ -15,11 +15,13 @@ internal sealed class SinglePartitionKeyServerBatchRequest : ServerBatchRequest
         /// Single partition key server request.
         /// </summary>
         /// <param name="partitionKey">Partition key that applies to all operations in this request.</param>
-        /// <param name="serializer">Serializer to serialize user provided objects to JSON.</param>
+        /// <param name="serializerCore">Serializer to serialize user provided objects to JSON.</param>
         private SinglePartitionKeyServerBatchRequest(
             PartitionKey? partitionKey,
-            CosmosSerializer serializer)
-            : base(maxBodyLength: int.MaxValue, maxOperationCount: int.MaxValue, serializer: serializer)
+            CosmosSerializerCore serializerCore)
+            : base(maxBodyLength: int.MaxValue,
+                  maxOperationCount: int.MaxValue,
+                  serializerCore: serializerCore)
         {
             this.PartitionKey = partitionKey;
         }
@@ -35,16 +37,16 @@ private SinglePartitionKeyServerBatchRequest(
         /// </summary>
         /// <param name="partitionKey">Partition key of the request.</param>
         /// <param name="operations">Operations to be added into this batch request.</param>
-        /// <param name="serializer">Serializer to serialize user provided objects to JSON.</param>
+        /// <param name="serializerCore">Serializer to serialize user provided objects to JSON.</param>
         /// <param name="cancellationToken"><see cref="CancellationToken"/> representing request cancellation.</param>
         /// <returns>A newly created instance of <see cref="SinglePartitionKeyServerBatchRequest"/>.</returns>
         public static async Task<SinglePartitionKeyServerBatchRequest> CreateAsync(
             PartitionKey? partitionKey,
             ArraySegment<ItemBatchOperation> operations,
-            CosmosSerializer serializer,
+            CosmosSerializerCore serializerCore,
             CancellationToken cancellationToken)
         {
-            SinglePartitionKeyServerBatchRequest request = new SinglePartitionKeyServerBatchRequest(partitionKey, serializer);
+            SinglePartitionKeyServerBatchRequest request = new SinglePartitionKeyServerBatchRequest(partitionKey, serializerCore);
             await request.CreateBodyStreamAsync(operations, cancellationToken);
             return request;
         }