Allow usage of ObjectSerializer for Cosmos bindings

extensions/Worker.Extensions.CosmosDB/release_notes.md

@@ -4,6 +4,6 @@
 - My change description (#PR/#issue)
 -->
 
-### Microsoft.Azure.Functions.Worker.Extensions.CosmosDB <4.14.0>
+### Microsoft.Azure.Functions.Worker.Extensions.CosmosDB <version>
 
-- Updates dependency `Microsoft.Azure.WebJobs.Extensions.CosmosDB` to version 4.11.0 (#3191)
+- Allow for specifying serializer for CosmosDB bindings (#3163)

extensions/Worker.Extensions.CosmosDB/src/Config/CosmosDBBindingOptions.cs

@@ -4,6 +4,7 @@
 using System;
 using System.Collections.Concurrent;
 using Azure.Core;
+using Azure.Core.Serialization;
 using Microsoft.Azure.Cosmos;
 using Microsoft.Azure.Functions.Worker.Extensions.CosmosDB;
 
@@ -17,7 +18,9 @@ namespace Microsoft.Azure.Functions.Worker
     /// </summary>
     internal class CosmosDBBindingOptions
     {
-        public string? ConnectionName  { get; set; }
+        private static readonly JsonObjectSerializer DefaultSerializer = new(new() { PropertyNameCaseInsensitive = true });
+
+        public string? ConnectionName { get; set; }
 
         public string? ConnectionString { get; set; }
 
@@ -27,6 +30,8 @@ internal class CosmosDBBindingOptions
 
         public CosmosDBExtensionOptions? CosmosExtensionOptions { get; set; }
 
+        public ObjectSerializer Serializer => CosmosExtensionOptions?.Serializer ?? DefaultSerializer;
+
         internal string BuildCacheKey(string connection, string region) => $"{connection}|{region}";
 
         internal ConcurrentDictionary<string, CosmosClient> ClientCache { get; } = new ConcurrentDictionary<string, CosmosClient>();

extensions/Worker.Extensions.CosmosDB/src/Config/CosmosDBBindingOptionsSetup.cs

@@ -2,7 +2,6 @@
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
 using System;
-using Microsoft.Azure.Cosmos;
 using Microsoft.Azure.Functions.Worker.Extensions;
 using Microsoft.Azure.Functions.Worker.Extensions.CosmosDB;
 using Microsoft.Extensions.Azure;
@@ -44,7 +43,6 @@ public void Configure(string? connectionName, CosmosDBBindingOptions options)
             }
 
             options.ConnectionName = connectionName;
-
             if (!string.IsNullOrWhiteSpace(connectionSection.Value))
             {
                 options.ConnectionString = connectionSection.Value;

extensions/Worker.Extensions.CosmosDB/src/Config/CosmosDBExtensionOptions.cs

@@ -1,6 +1,7 @@
 // Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
+using Azure.Core.Serialization;
 using Microsoft.Azure.Cosmos;
 
 namespace Microsoft.Azure.Functions.Worker
@@ -11,5 +12,8 @@ public class CosmosDBExtensionOptions
         /// Gets or sets the CosmosClientOptions.
         /// </summary>
         public CosmosClientOptions ClientOptions { get; set; } = new() { ConnectionMode = ConnectionMode.Gateway };
+
+        // TODO: in next major version, ensure this is WorkerOptions.Serializer by default.
+        public ObjectSerializer? Serializer { get; set; }
     }
 }

extensions/Worker.Extensions.CosmosDB/src/CosmosDBConverter.cs

@@ -6,7 +6,6 @@
 using System.Collections.Generic;
 using System.IO;
 using System.Linq;
-using System.Text.Json;
 using System.Threading.Tasks;
 using Microsoft.Azure.Cosmos;
 using Microsoft.Azure.Functions.Worker.Core;
@@ -16,6 +15,7 @@
 using Microsoft.Extensions.Logging;
 using Microsoft.Azure.Functions.Worker.Extensions;
 using Microsoft.Azure.Functions.Worker.Extensions.Abstractions;
+using Azure.Core.Serialization;
 
 namespace Microsoft.Azure.Functions.Worker
 {
@@ -27,9 +27,9 @@ internal class CosmosDBConverter : IInputConverter
     {
         private readonly IOptionsMonitor<CosmosDBBindingOptions> _cosmosOptions;
         private readonly ILogger<CosmosDBConverter> _logger;
-        private static readonly JsonSerializerOptions _serializerOptions = new() { PropertyNameCaseInsensitive = true };
 
-        public CosmosDBConverter(IOptionsMonitor<CosmosDBBindingOptions> cosmosOptions, ILogger<CosmosDBConverter> logger)
+        public CosmosDBConverter(
+            IOptionsMonitor<CosmosDBBindingOptions> cosmosOptions, ILogger<CosmosDBConverter> logger)
         {
             _cosmosOptions = cosmosOptions ?? throw new ArgumentNullException(nameof(cosmosOptions));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
@@ -44,7 +44,8 @@ public async ValueTask<ConversionResult> ConvertAsync(ConverterContext context)
             };
         }
 
-        private async ValueTask<ConversionResult> ConvertFromBindingDataAsync(ConverterContext context, ModelBindingData modelBindingData)
+        private async ValueTask<ConversionResult> ConvertFromBindingDataAsync(
+            ConverterContext context, ModelBindingData modelBindingData)
         {
             try
             {
@@ -55,7 +56,6 @@ private async ValueTask<ConversionResult> ConvertFromBindingDataAsync(ConverterC
 
                 CosmosDBInputAttribute cosmosAttribute = GetBindingDataContent(modelBindingData);
                 object result = await ToTargetTypeAsync(context.TargetType, cosmosAttribute);
-
                 return ConversionResult.Success(result);
             }
             catch (CosmosException ex) when (ex.StatusCode == System.Net.HttpStatusCode.NotFound)
@@ -83,60 +83,75 @@ private CosmosDBInputAttribute GetBindingDataContent(ModelBindingData bindingDat
             };
         }
 
-        private async Task<object> ToTargetTypeAsync(Type targetType, CosmosDBInputAttribute cosmosAttribute) => targetType switch
-        {
-            Type _ when targetType == typeof(CosmosClient) => CreateCosmosClient<CosmosClient>(cosmosAttribute),
-            Type _ when targetType == typeof(Database) => CreateCosmosClient<Database>(cosmosAttribute),
-            Type _ when targetType == typeof(Container) => CreateCosmosClient<Container>(cosmosAttribute),
-            _ => await CreateTargetObjectAsync(targetType, cosmosAttribute)
-        };
+        private async Task<object> ToTargetTypeAsync(Type targetType, CosmosDBInputAttribute cosmosAttribute)
+            => targetType switch
+            {
+                Type _ when targetType == typeof(CosmosClient) => CreateCosmosClient<CosmosClient>(cosmosAttribute),
+                Type _ when targetType == typeof(Database) => CreateCosmosClient<Database>(cosmosAttribute),
+                Type _ when targetType == typeof(Container) => CreateCosmosClient<Container>(cosmosAttribute),
+                _ => await CreateTargetObjectAsync(targetType, cosmosAttribute)
+            };
 
         private async Task<object> CreateTargetObjectAsync(Type targetType, CosmosDBInputAttribute cosmosAttribute)
         {
-            if (CreateCosmosClient<Container>(cosmosAttribute) is not Container container)
+            if (CreateCosmosClient<Container>(cosmosAttribute, out ObjectSerializer serializer)
+                is not Container container)
             {
-                throw new InvalidOperationException($"Unable to create Cosmos container client for '{cosmosAttribute.ContainerName}'.");
+                throw new InvalidOperationException(
+                    $"Unable to create Cosmos container client for '{cosmosAttribute.ContainerName}'.");
             }
 
             if (targetType.IsCollectionType())
             {
                 return await ParameterBinder.BindCollectionAsync(
-                    elementType => GetDocumentsAsync(container, cosmosAttribute, elementType), targetType);
+                    elementType => GetDocumentsAsync(container, serializer, cosmosAttribute, elementType), targetType);
             }
             else
             {
-                return await CreatePocoAsync(container, cosmosAttribute, targetType);
+                return await CreatePocoAsync(container, serializer, cosmosAttribute, targetType);
             }
         }
 
-        private async Task<object> CreatePocoAsync(Container container, CosmosDBInputAttribute cosmosAttribute, Type targetType)
+        private async Task<object> CreatePocoAsync(
+            Container container,
+            ObjectSerializer serializer,
+            CosmosDBInputAttribute cosmosAttribute,
+            Type targetType)
         {
             if (string.IsNullOrEmpty(cosmosAttribute.Id) || string.IsNullOrEmpty(cosmosAttribute.PartitionKey))
             {
-                throw new InvalidOperationException("The 'Id' and 'PartitionKey' properties of a CosmosDB single-item input binding cannot be null or empty.");
+                throw new InvalidOperationException(
+                    "The 'Id' and 'PartitionKey' properties of a CosmosDB single-item input binding cannot be null or empty.");
             }
 
-            ResponseMessage item = await container.ReadItemStreamAsync(cosmosAttribute.Id, new PartitionKey(cosmosAttribute.PartitionKey));
+            using ResponseMessage item = await container.ReadItemStreamAsync(
+                cosmosAttribute.Id, new PartitionKey(cosmosAttribute.PartitionKey));
             item.EnsureSuccessStatusCode();
-            return (await JsonSerializer.DeserializeAsync(item.Content, targetType, _serializerOptions))!;
+
+            return (await serializer.DeserializeAsync(item.Content, targetType, default))!;
         }
 
-        private async IAsyncEnumerable<object> GetDocumentsAsync(Container container, CosmosDBInputAttribute cosmosAttribute, Type elementType)
+        private async IAsyncEnumerable<object> GetDocumentsAsync(
+            Container container,
+            ObjectSerializer serializer,
+            CosmosDBInputAttribute cosmosAttribute,
+            Type elementType)
         {
             await foreach (var stream in GetDocumentsStreamAsync(container, cosmosAttribute))
             {
-                // Cosmos returns a stream of JSON which represents a paged response. The contents are in a property called "Documents".
-                // Deserializing into CosmosStream<T> will extract these documents.
+                // Cosmos returns a stream of JSON which represents a paged response. The contents are in a
+                // property called "Documents". Deserializing into CosmosStream<T> will extract these documents.
                 Type target = typeof(CosmosStream<>).MakeGenericType(elementType);
-                CosmosStream page = (CosmosStream)(await JsonSerializer.DeserializeAsync(stream!, target, _serializerOptions))!;
+                CosmosStream page = (CosmosStream)(await serializer.DeserializeAsync(stream!, target, default))!;
                 foreach (var item in page.GetDocuments())
                 {
                     yield return item;
                 }
             }
         }
 
-        private async IAsyncEnumerable<Stream> GetDocumentsStreamAsync(Container container, CosmosDBInputAttribute cosmosAttribute)
+        private async IAsyncEnumerable<Stream> GetDocumentsStreamAsync(
+            Container container, CosmosDBInputAttribute cosmosAttribute)
         {
             QueryDefinition queryDefinition = null!;
             if (!string.IsNullOrEmpty(cosmosAttribute.SqlQuery))
@@ -158,8 +173,10 @@ private async IAsyncEnumerable<Stream> GetDocumentsStreamAsync(Container contain
                 queryRequestOptions = new() { PartitionKey = partitionKey };
             }
 
-            using FeedIterator iterator = container.GetItemQueryStreamIterator(queryDefinition: queryDefinition, requestOptions: queryRequestOptions)
-                ?? throw new InvalidOperationException($"Unable to retrieve documents for container '{container.Id}'.");
+            using FeedIterator iterator = container.GetItemQueryStreamIterator(
+                queryDefinition: queryDefinition, requestOptions: queryRequestOptions)
+                ?? throw new InvalidOperationException(
+                    $"Unable to retrieve documents for container '{container.Id}'.");
 
             while (iterator.HasMoreResults)
             {
@@ -170,23 +187,28 @@ private async IAsyncEnumerable<Stream> GetDocumentsStreamAsync(Container contain
         }
 
         private T CreateCosmosClient<T>(CosmosDBInputAttribute cosmosAttribute)
+            => CreateCosmosClient<T>(cosmosAttribute, out _);
+
+        private T CreateCosmosClient<T>(CosmosDBInputAttribute cosmosAttribute, out ObjectSerializer serializer)
         {
             if (cosmosAttribute is null)
             {
                 throw new ArgumentNullException(nameof(cosmosAttribute));
             }
 
-            var cosmosDBOptions = _cosmosOptions.Get(cosmosAttribute.Connection);
+            CosmosDBBindingOptions cosmosDBOptions = _cosmosOptions.Get(cosmosAttribute.Connection);
             CosmosClient cosmosClient = cosmosDBOptions.GetClient(cosmosAttribute.PreferredLocations!);
 
             Type targetType = typeof(T);
             object cosmosReference = targetType switch
             {
                 Type _ when targetType == typeof(Database) => cosmosClient.GetDatabase(cosmosAttribute.DatabaseName),
-                Type _ when targetType == typeof(Container) => cosmosClient.GetContainer(cosmosAttribute.DatabaseName, cosmosAttribute.ContainerName),
+                Type _ when targetType == typeof(Container) => cosmosClient.GetContainer(
+                    cosmosAttribute.DatabaseName, cosmosAttribute.ContainerName),
                 _ => cosmosClient
             };
 
+            serializer = cosmosDBOptions.Serializer;
             return (T)cosmosReference;
         }
 

extensions/Worker.Extensions.CosmosDB/src/Extensions/FunctionsWorkerApplicationBuilderExtensions.cs

@@ -2,7 +2,7 @@
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
 using System;
-using Microsoft.Azure.Cosmos;
+using Azure.Core.Serialization;
 using Microsoft.Extensions.Azure;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
@@ -20,25 +20,28 @@ public static class FunctionsWorkerApplicationBuilderExtensions
         /// </summary>
         /// <param name="builder">The <see cref="IFunctionsWorkerApplicationBuilder"/> to configure.</param>
         /// <returns>The same instance of the <see cref="IFunctionsWorkerApplicationBuilder"/> for chaining.</returns>
-        public static IFunctionsWorkerApplicationBuilder ConfigureCosmosDBExtension(this IFunctionsWorkerApplicationBuilder builder)
+        public static IFunctionsWorkerApplicationBuilder ConfigureCosmosDBExtension(
+            this IFunctionsWorkerApplicationBuilder builder)
         {
             if (builder is null)
             {
-                throw new System.ArgumentNullException(nameof(builder));
+                throw new ArgumentNullException(nameof(builder));
             }
 
             builder.Services.AddAzureClientsCore(); // Adds AzureComponentFactory
             builder.Services.AddOptions<CosmosDBBindingOptions>();
             builder.Services.AddOptions<CosmosDBExtensionOptions>()
-                            .Configure<IOptions<WorkerOptions>>((cosmos, worker) =>
-                            {
-                                if (cosmos.ClientOptions.Serializer is null)
-                                {
-                                    cosmos.ClientOptions.Serializer = new WorkerCosmosSerializer(worker.Value.Serializer);
-                                }
-                            });
+                .PostConfigure<IOptions<WorkerOptions>>((cosmos, worker) =>
+                {
+                    ObjectSerializer? serializer = cosmos.Serializer ?? worker.Value.Serializer;
+                    if (serializer is not null && cosmos.ClientOptions.Serializer is null)
+                    {
+                        cosmos.ClientOptions.Serializer = new WorkerCosmosSerializer(serializer);
+                    }
+                });
 
-            builder.Services.TryAddEnumerable(ServiceDescriptor.Singleton<IConfigureOptions<CosmosDBBindingOptions>, CosmosDBBindingOptionsSetup>());
+            builder.Services.TryAddEnumerable(ServiceDescriptor.Singleton<
+                IConfigureOptions<CosmosDBBindingOptions>, CosmosDBBindingOptionsSetup>());
 
             return builder;
         }
@@ -49,10 +52,27 @@ public static IFunctionsWorkerApplicationBuilder ConfigureCosmosDBExtension(this
         /// <param name="builder">The IFunctionsWorkerApplicationBuilder to add the configuration to.</param>
         /// <param name="options">An Action to configure the CosmosDBExtensionOptions.</param>
         /// <returns>The same instance of the <see cref="IFunctionsWorkerApplicationBuilder"/> for chaining.</returns>
-        public static IFunctionsWorkerApplicationBuilder ConfigureCosmosDBExtensionOptions(this IFunctionsWorkerApplicationBuilder builder, Action<CosmosDBExtensionOptions> options)
+        public static IFunctionsWorkerApplicationBuilder ConfigureCosmosDBExtensionOptions(
+            this IFunctionsWorkerApplicationBuilder builder, Action<CosmosDBExtensionOptions> options)
         {
             builder.Services.Configure(options);
             return builder;
         }
+
+        /// <summary>
+        /// Configures the CosmosDBExtensionOptions for the Functions Worker Cosmos extension.
+        /// </summary>
+        /// <param name="builder">The IFunctionsWorkerApplicationBuilder to add the configuration to.</param>
+        /// <returns>The same instance of the <see cref="IFunctionsWorkerApplicationBuilder"/> for chaining.</returns>
+        public static IFunctionsWorkerApplicationBuilder UseCosmosDBWorkerSerializer(this IFunctionsWorkerApplicationBuilder builder)
+        {
+            builder.Services.AddOptions<CosmosDBExtensionOptions>()
+                .Configure<IOptions<WorkerOptions>>((cosmos, worker) =>
+                {
+                    cosmos.Serializer ??= worker.Value.Serializer;
+                });
+
+            return builder;
+        }
     }
 }

extensions/Worker.Extensions.CosmosDB/src/WorkerCosmosSerializer.cs

@@ -33,13 +33,13 @@ public WorkerCosmosSerializer(ObjectSerializer? serializer)
         /// <returns>The object representing the deserialized stream.</returns>
         public override T FromStream<T>(Stream stream)
         {
-            using (stream)
+            if (typeof(Stream).IsAssignableFrom(typeof(T)))
             {
-                if (typeof(Stream).IsAssignableFrom(typeof(T)))
-                {
-                    return (T)(object)stream;
-                }
+                return (T)(object)stream;
+            }
 
+            using (stream)
+            {
                 return (T)_serializer.Deserialize(stream, typeof(T), default)!;
             }
         }
@@ -52,10 +52,10 @@ public override T FromStream<T>(Stream stream)
         /// <returns>An open readable stream containing the JSON of the serialized object.</returns>
         public override Stream ToStream<T>(T input)
         {
-            var streamPayload = new MemoryStream();
+            MemoryStream streamPayload = new();
             _serializer.Serialize(streamPayload, input, typeof(T), default);
             streamPayload.Position = 0;
             return streamPayload;
         }
     }
-}
+}