diff --git a/extensions/Worker.Extensions.CosmosDB/release_notes.md b/extensions/Worker.Extensions.CosmosDB/release_notes.md index b9c7c90b7..bc65d6761 100644 --- a/extensions/Worker.Extensions.CosmosDB/release_notes.md +++ b/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 -- Updates dependency `Microsoft.Azure.WebJobs.Extensions.CosmosDB` to version 4.11.0 (#3191) +- Allow for specifying serializer for CosmosDB bindings (#3163) diff --git a/extensions/Worker.Extensions.CosmosDB/src/Config/CosmosDBBindingOptions.cs b/extensions/Worker.Extensions.CosmosDB/src/Config/CosmosDBBindingOptions.cs index 403ee836a..438825411 100644 --- a/extensions/Worker.Extensions.CosmosDB/src/Config/CosmosDBBindingOptions.cs +++ b/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 /// 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 ClientCache { get; } = new ConcurrentDictionary(); diff --git a/extensions/Worker.Extensions.CosmosDB/src/Config/CosmosDBBindingOptionsSetup.cs b/extensions/Worker.Extensions.CosmosDB/src/Config/CosmosDBBindingOptionsSetup.cs index 4fa6b5375..aca003f39 100644 --- a/extensions/Worker.Extensions.CosmosDB/src/Config/CosmosDBBindingOptionsSetup.cs +++ b/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; diff --git a/extensions/Worker.Extensions.CosmosDB/src/Config/CosmosDBExtensionOptions.cs b/extensions/Worker.Extensions.CosmosDB/src/Config/CosmosDBExtensionOptions.cs index e9d7b3c75..12713c4de 100644 --- a/extensions/Worker.Extensions.CosmosDB/src/Config/CosmosDBExtensionOptions.cs +++ b/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. /// 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; } } } diff --git a/extensions/Worker.Extensions.CosmosDB/src/CosmosDBConverter.cs b/extensions/Worker.Extensions.CosmosDB/src/CosmosDBConverter.cs index 5a7331543..8b894f3ae 100644 --- a/extensions/Worker.Extensions.CosmosDB/src/CosmosDBConverter.cs +++ b/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 _cosmosOptions; private readonly ILogger _logger; - private static readonly JsonSerializerOptions _serializerOptions = new() { PropertyNameCaseInsensitive = true }; - public CosmosDBConverter(IOptionsMonitor cosmosOptions, ILogger logger) + public CosmosDBConverter( + IOptionsMonitor cosmosOptions, ILogger logger) { _cosmosOptions = cosmosOptions ?? throw new ArgumentNullException(nameof(cosmosOptions)); _logger = logger ?? throw new ArgumentNullException(nameof(logger)); @@ -44,7 +44,8 @@ public async ValueTask ConvertAsync(ConverterContext context) }; } - private async ValueTask ConvertFromBindingDataAsync(ConverterContext context, ModelBindingData modelBindingData) + private async ValueTask ConvertFromBindingDataAsync( + ConverterContext context, ModelBindingData modelBindingData) { try { @@ -55,7 +56,6 @@ private async ValueTask 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,52 +83,66 @@ private CosmosDBInputAttribute GetBindingDataContent(ModelBindingData bindingDat }; } - private async Task ToTargetTypeAsync(Type targetType, CosmosDBInputAttribute cosmosAttribute) => targetType switch - { - Type _ when targetType == typeof(CosmosClient) => CreateCosmosClient(cosmosAttribute), - Type _ when targetType == typeof(Database) => CreateCosmosClient(cosmosAttribute), - Type _ when targetType == typeof(Container) => CreateCosmosClient(cosmosAttribute), - _ => await CreateTargetObjectAsync(targetType, cosmosAttribute) - }; + private async Task ToTargetTypeAsync(Type targetType, CosmosDBInputAttribute cosmosAttribute) + => targetType switch + { + Type _ when targetType == typeof(CosmosClient) => CreateCosmosClient(cosmosAttribute), + Type _ when targetType == typeof(Database) => CreateCosmosClient(cosmosAttribute), + Type _ when targetType == typeof(Container) => CreateCosmosClient(cosmosAttribute), + _ => await CreateTargetObjectAsync(targetType, cosmosAttribute) + }; private async Task CreateTargetObjectAsync(Type targetType, CosmosDBInputAttribute cosmosAttribute) { - if (CreateCosmosClient(cosmosAttribute) is not Container container) + if (CreateCosmosClient(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 CreatePocoAsync(Container container, CosmosDBInputAttribute cosmosAttribute, Type targetType) + private async Task 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 GetDocumentsAsync(Container container, CosmosDBInputAttribute cosmosAttribute, Type elementType) + private async IAsyncEnumerable 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 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 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; @@ -136,7 +150,8 @@ private async IAsyncEnumerable GetDocumentsAsync(Container container, Co } } - private async IAsyncEnumerable GetDocumentsStreamAsync(Container container, CosmosDBInputAttribute cosmosAttribute) + private async IAsyncEnumerable GetDocumentsStreamAsync( + Container container, CosmosDBInputAttribute cosmosAttribute) { QueryDefinition queryDefinition = null!; if (!string.IsNullOrEmpty(cosmosAttribute.SqlQuery)) @@ -158,8 +173,10 @@ private async IAsyncEnumerable 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 GetDocumentsStreamAsync(Container contain } private T CreateCosmosClient(CosmosDBInputAttribute cosmosAttribute) + => CreateCosmosClient(cosmosAttribute, out _); + + private T CreateCosmosClient(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; } diff --git a/extensions/Worker.Extensions.CosmosDB/src/Extensions/FunctionsWorkerApplicationBuilderExtensions.cs b/extensions/Worker.Extensions.CosmosDB/src/Extensions/FunctionsWorkerApplicationBuilderExtensions.cs index dc8743bbd..c40591eb1 100644 --- a/extensions/Worker.Extensions.CosmosDB/src/Extensions/FunctionsWorkerApplicationBuilderExtensions.cs +++ b/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 /// /// The to configure. /// The same instance of the for chaining. - 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(); builder.Services.AddOptions() - .Configure>((cosmos, worker) => - { - if (cosmos.ClientOptions.Serializer is null) - { - cosmos.ClientOptions.Serializer = new WorkerCosmosSerializer(worker.Value.Serializer); - } - }); + .PostConfigure>((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, CosmosDBBindingOptionsSetup>()); + builder.Services.TryAddEnumerable(ServiceDescriptor.Singleton< + IConfigureOptions, CosmosDBBindingOptionsSetup>()); return builder; } @@ -49,10 +52,27 @@ public static IFunctionsWorkerApplicationBuilder ConfigureCosmosDBExtension(this /// The IFunctionsWorkerApplicationBuilder to add the configuration to. /// An Action to configure the CosmosDBExtensionOptions. /// The same instance of the for chaining. - public static IFunctionsWorkerApplicationBuilder ConfigureCosmosDBExtensionOptions(this IFunctionsWorkerApplicationBuilder builder, Action options) + public static IFunctionsWorkerApplicationBuilder ConfigureCosmosDBExtensionOptions( + this IFunctionsWorkerApplicationBuilder builder, Action options) { builder.Services.Configure(options); return builder; } + + /// + /// Configures the CosmosDBExtensionOptions for the Functions Worker Cosmos extension. + /// + /// The IFunctionsWorkerApplicationBuilder to add the configuration to. + /// The same instance of the for chaining. + public static IFunctionsWorkerApplicationBuilder UseCosmosDBWorkerSerializer(this IFunctionsWorkerApplicationBuilder builder) + { + builder.Services.AddOptions() + .Configure>((cosmos, worker) => + { + cosmos.Serializer ??= worker.Value.Serializer; + }); + + return builder; + } } } diff --git a/extensions/Worker.Extensions.CosmosDB/src/WorkerCosmosSerializer.cs b/extensions/Worker.Extensions.CosmosDB/src/WorkerCosmosSerializer.cs index 480f16d05..569257680 100644 --- a/extensions/Worker.Extensions.CosmosDB/src/WorkerCosmosSerializer.cs +++ b/extensions/Worker.Extensions.CosmosDB/src/WorkerCosmosSerializer.cs @@ -33,13 +33,13 @@ public WorkerCosmosSerializer(ObjectSerializer? serializer) /// The object representing the deserialized stream. public override T FromStream(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(Stream stream) /// An open readable stream containing the JSON of the serialized object. public override Stream ToStream(T input) { - var streamPayload = new MemoryStream(); + MemoryStream streamPayload = new(); _serializer.Serialize(streamPayload, input, typeof(T), default); streamPayload.Position = 0; return streamPayload; } } -} \ No newline at end of file +}