From 5699e34ff4a6f7e75e01152bc94f2b721a049e58 Mon Sep 17 00:00:00 2001 From: JoshLove-msft <54595583+JoshLove-msft@users.noreply.github.com> Date: Mon, 29 Nov 2021 12:13:15 -0800 Subject: [PATCH 1/6] save --- eng/Packages.Data.props | 1 - .../Azure.Core.Experimental.netstandard2.0.cs | 8 + .../src/IMessageWithContentType.cs | 23 ++ .../tests/ConsumeEventTests.cs | 2 +- .../Azure.Messaging.EventHubs.Shared.sln | 28 ++ ...re.Messaging.EventHubs.Shared.Tests.csproj | 1 + .../src/Azure.Messaging.EventHubs.csproj | 5 +- .../src/EventData.cs | 13 +- ...t.Azure.Data.SchemaRegistry.ApacheAvro.sln | 12 + .../README.md | 47 +-- ...chemaRegistry.ApacheAvro.netstandard2.0.cs | 16 +- ...zure.Data.SchemaRegistry.ApacheAvro.csproj | 6 +- .../src/Properties/AssemblyInfo.cs | 6 + .../src/SchemaRegistryAvroEncoder.cs | 331 ++++++++++++++++++ ...SchemaRegistryAvroObjectEncoderOptions.cs} | 4 +- .../src/SchemaRegistryAvroObjectSerializer.cs | 261 -------------- ...ata.SchemaRegistry.ApacheAvro.Tests.csproj | 2 + .../tests/Samples/Sample01_ReadmeSnippets.cs | 59 ++-- ...maRegistryAvroObjectSerializerLiveTests.cs | 67 +++- .../CanUseEncoderWithEventData.json | 52 +++ .../CanUseEncoderWithEventDataAsync.json | 52 +++ .../CannotDeserializeWithNullSchemaId.json | 8 + ...annotDeserializeWithNullSchemaIdAsync.json | 8 + .../Administration/RulePropertiesTests.cs | 2 +- 24 files changed, 649 insertions(+), 365 deletions(-) create mode 100644 sdk/core/Azure.Core.Experimental/src/IMessageWithContentType.cs create mode 100644 sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/Properties/AssemblyInfo.cs create mode 100644 sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroEncoder.cs rename sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/{SchemaRegistryAvroObjectSerializerOptions.cs => SchemaRegistryAvroObjectEncoderOptions.cs} (82%) delete mode 100644 sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroObjectSerializer.cs create mode 100644 sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithEventDataAsync.json create mode 100644 sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CannotDeserializeWithNullSchemaId.json create mode 100644 sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CannotDeserializeWithNullSchemaIdAsync.json diff --git a/eng/Packages.Data.props b/eng/Packages.Data.props index cf5879b8d5e37..eb83a067a946b 100644 --- a/eng/Packages.Data.props +++ b/eng/Packages.Data.props @@ -82,7 +82,6 @@ - diff --git a/sdk/core/Azure.Core.Experimental/api/Azure.Core.Experimental.netstandard2.0.cs b/sdk/core/Azure.Core.Experimental/api/Azure.Core.Experimental.netstandard2.0.cs index e953b9400fadb..da37425ab060a 100644 --- a/sdk/core/Azure.Core.Experimental/api/Azure.Core.Experimental.netstandard2.0.cs +++ b/sdk/core/Azure.Core.Experimental/api/Azure.Core.Experimental.netstandard2.0.cs @@ -115,3 +115,11 @@ public static partial class ResponseExtensions public static bool IsError(this Azure.Response response) { throw null; } } } +namespace Azure.Messaging +{ + public partial interface IMessageWithContentType + { + string ContentType { get; set; } + System.BinaryData Data { get; set; } + } +} diff --git a/sdk/core/Azure.Core.Experimental/src/IMessageWithContentType.cs b/sdk/core/Azure.Core.Experimental/src/IMessageWithContentType.cs new file mode 100644 index 0000000000000..1c895c3fe0a0e --- /dev/null +++ b/sdk/core/Azure.Core.Experimental/src/IMessageWithContentType.cs @@ -0,0 +1,23 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +using System; + +namespace Azure.Messaging +{ + /// + /// An interface for a message containing a content type along with its data. + /// + public interface IMessageWithContentType + { + /// + /// Gets or sets the message data. + /// + BinaryData Data { get; set; } + + /// + /// Gets or sets the message content type. + /// + string ContentType { get; set; } + } +} \ No newline at end of file diff --git a/sdk/eventgrid/Azure.Messaging.EventGrid/tests/ConsumeEventTests.cs b/sdk/eventgrid/Azure.Messaging.EventGrid/tests/ConsumeEventTests.cs index b582307c48a87..3fb01d602cbbd 100644 --- a/sdk/eventgrid/Azure.Messaging.EventGrid/tests/ConsumeEventTests.cs +++ b/sdk/eventgrid/Azure.Messaging.EventGrid/tests/ConsumeEventTests.cs @@ -447,7 +447,7 @@ public void ConsumeEventHubCaptureFileCreatedEvent() Assert.NotNull(events); Assert.True(events[0].TryGetSystemEventData(out object eventData)); - Assert.AreEqual("AzureBlockBlob", (eventData as EventHubCaptureFileCreatedEventData).FileType); + Assert.AreEqual("AzureBlockBlob", (eventData as EventHubCaptureFileCreatedEventData).Fileurl); } #endregion diff --git a/sdk/eventhub/Azure.Messaging.EventHubs.Shared/Azure.Messaging.EventHubs.Shared.sln b/sdk/eventhub/Azure.Messaging.EventHubs.Shared/Azure.Messaging.EventHubs.Shared.sln index 77f40f669b946..e823fd0bf6a77 100755 --- a/sdk/eventhub/Azure.Messaging.EventHubs.Shared/Azure.Messaging.EventHubs.Shared.sln +++ b/sdk/eventhub/Azure.Messaging.EventHubs.Shared/Azure.Messaging.EventHubs.Shared.sln @@ -13,6 +13,10 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "External", "External", "{10 EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Azure.Messaging.EventHubs", "..\Azure.Messaging.EventHubs\src\Azure.Messaging.EventHubs.csproj", "{870D6ACA-B778-40AA-93B5-43CF1D25235E}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Azure.Core.Experimental", "..\..\core\Azure.Core.Experimental\src\Azure.Core.Experimental.csproj", "{07A39A33-E454-4404-85DC-42B603AEBD6D}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Azure.Core", "..\..\core\Azure.Core\src\Azure.Core.csproj", "{E3B14DB9-48E1-4B7D-A384-3247163A1421}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -59,6 +63,30 @@ Global {870D6ACA-B778-40AA-93B5-43CF1D25235E}.Release|x64.Build.0 = Release|Any CPU {870D6ACA-B778-40AA-93B5-43CF1D25235E}.Release|x86.ActiveCfg = Release|Any CPU {870D6ACA-B778-40AA-93B5-43CF1D25235E}.Release|x86.Build.0 = Release|Any CPU + {07A39A33-E454-4404-85DC-42B603AEBD6D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {07A39A33-E454-4404-85DC-42B603AEBD6D}.Debug|Any CPU.Build.0 = Debug|Any CPU + {07A39A33-E454-4404-85DC-42B603AEBD6D}.Debug|x64.ActiveCfg = Debug|Any CPU + {07A39A33-E454-4404-85DC-42B603AEBD6D}.Debug|x64.Build.0 = Debug|Any CPU + {07A39A33-E454-4404-85DC-42B603AEBD6D}.Debug|x86.ActiveCfg = Debug|Any CPU + {07A39A33-E454-4404-85DC-42B603AEBD6D}.Debug|x86.Build.0 = Debug|Any CPU + {07A39A33-E454-4404-85DC-42B603AEBD6D}.Release|Any CPU.ActiveCfg = Release|Any CPU + {07A39A33-E454-4404-85DC-42B603AEBD6D}.Release|Any CPU.Build.0 = Release|Any CPU + {07A39A33-E454-4404-85DC-42B603AEBD6D}.Release|x64.ActiveCfg = Release|Any CPU + {07A39A33-E454-4404-85DC-42B603AEBD6D}.Release|x64.Build.0 = Release|Any CPU + {07A39A33-E454-4404-85DC-42B603AEBD6D}.Release|x86.ActiveCfg = Release|Any CPU + {07A39A33-E454-4404-85DC-42B603AEBD6D}.Release|x86.Build.0 = Release|Any CPU + {E3B14DB9-48E1-4B7D-A384-3247163A1421}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {E3B14DB9-48E1-4B7D-A384-3247163A1421}.Debug|Any CPU.Build.0 = Debug|Any CPU + {E3B14DB9-48E1-4B7D-A384-3247163A1421}.Debug|x64.ActiveCfg = Debug|Any CPU + {E3B14DB9-48E1-4B7D-A384-3247163A1421}.Debug|x64.Build.0 = Debug|Any CPU + {E3B14DB9-48E1-4B7D-A384-3247163A1421}.Debug|x86.ActiveCfg = Debug|Any CPU + {E3B14DB9-48E1-4B7D-A384-3247163A1421}.Debug|x86.Build.0 = Debug|Any CPU + {E3B14DB9-48E1-4B7D-A384-3247163A1421}.Release|Any CPU.ActiveCfg = Release|Any CPU + {E3B14DB9-48E1-4B7D-A384-3247163A1421}.Release|Any CPU.Build.0 = Release|Any CPU + {E3B14DB9-48E1-4B7D-A384-3247163A1421}.Release|x64.ActiveCfg = Release|Any CPU + {E3B14DB9-48E1-4B7D-A384-3247163A1421}.Release|x64.Build.0 = Release|Any CPU + {E3B14DB9-48E1-4B7D-A384-3247163A1421}.Release|x86.ActiveCfg = Release|Any CPU + {E3B14DB9-48E1-4B7D-A384-3247163A1421}.Release|x86.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE diff --git a/sdk/eventhub/Azure.Messaging.EventHubs.Shared/tests/Azure.Messaging.EventHubs.Shared.Tests.csproj b/sdk/eventhub/Azure.Messaging.EventHubs.Shared/tests/Azure.Messaging.EventHubs.Shared.Tests.csproj index bf55719e4b75f..407d194fb7e94 100644 --- a/sdk/eventhub/Azure.Messaging.EventHubs.Shared/tests/Azure.Messaging.EventHubs.Shared.Tests.csproj +++ b/sdk/eventhub/Azure.Messaging.EventHubs.Shared/tests/Azure.Messaging.EventHubs.Shared.Tests.csproj @@ -27,6 +27,7 @@ + diff --git a/sdk/eventhub/Azure.Messaging.EventHubs/src/Azure.Messaging.EventHubs.csproj b/sdk/eventhub/Azure.Messaging.EventHubs/src/Azure.Messaging.EventHubs.csproj index 09f2a8241021c..a0b874090cd67 100644 --- a/sdk/eventhub/Azure.Messaging.EventHubs/src/Azure.Messaging.EventHubs.csproj +++ b/sdk/eventhub/Azure.Messaging.EventHubs/src/Azure.Messaging.EventHubs.csproj @@ -14,7 +14,7 @@ - + @@ -59,4 +59,7 @@ Azure.Messaging.EventHubs + + + diff --git a/sdk/eventhub/Azure.Messaging.EventHubs/src/EventData.cs b/sdk/eventhub/Azure.Messaging.EventHubs/src/EventData.cs index 801250f188215..867678859df3b 100644 --- a/sdk/eventhub/Azure.Messaging.EventHubs/src/EventData.cs +++ b/sdk/eventhub/Azure.Messaging.EventHubs/src/EventData.cs @@ -7,8 +7,8 @@ using System.IO; using System.Text; using Azure.Core; -using Azure.Core.Amqp; using Azure.Core.Serialization; +using Azure.Core.Amqp; using Azure.Messaging.EventHubs.Amqp; using Azure.Messaging.EventHubs.Consumer; using Azure.Messaging.EventHubs.Producer; @@ -19,7 +19,7 @@ namespace Azure.Messaging.EventHubs /// An Event Hubs event, encapsulating a set of data and its associated metadata. /// /// - public class EventData + public class EventData : IMessageWithContentType { /// The AMQP representation of the event, allowing access to additional protocol data elements not used directly by the Event Hubs client library. private readonly AmqpAnnotatedMessage _amqpMessage; @@ -48,6 +48,15 @@ public BinaryData EventBody set => _amqpMessage.Body = AmqpMessageBody.FromData(MessageBody.FromReadOnlyMemorySegment(value.ToMemory())); } + /// + /// + /// + public BinaryData Data + { + get => EventBody; + set => EventBody = value; + } + /// /// A MIME type describing the data contained in the , /// intended to allow consumers to make informed decisions for inspecting and diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.sln b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.sln index afe1df921037d..8644d54c1f449 100644 --- a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.sln +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.sln @@ -11,6 +11,10 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Azure.Core.TestFramework", EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Azure.Data.SchemaRegistry", "..\Azure.Data.SchemaRegistry\src\Azure.Data.SchemaRegistry.csproj", "{D451EE68-ADE4-4780-A002-2D13DEA888A2}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Azure.Core.Experimental", "..\..\core\Azure.Core.Experimental\src\Azure.Core.Experimental.csproj", "{8A0A0FBB-2B12-49B3-951E-5402D64075DD}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Azure.Messaging.EventHubs", "..\..\eventhub\Azure.Messaging.EventHubs\src\Azure.Messaging.EventHubs.csproj", "{FDC382B3-7255-4D33-AAA9-1168B7D01E94}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -33,6 +37,14 @@ Global {D451EE68-ADE4-4780-A002-2D13DEA888A2}.Debug|Any CPU.Build.0 = Debug|Any CPU {D451EE68-ADE4-4780-A002-2D13DEA888A2}.Release|Any CPU.ActiveCfg = Release|Any CPU {D451EE68-ADE4-4780-A002-2D13DEA888A2}.Release|Any CPU.Build.0 = Release|Any CPU + {8A0A0FBB-2B12-49B3-951E-5402D64075DD}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {8A0A0FBB-2B12-49B3-951E-5402D64075DD}.Debug|Any CPU.Build.0 = Debug|Any CPU + {8A0A0FBB-2B12-49B3-951E-5402D64075DD}.Release|Any CPU.ActiveCfg = Release|Any CPU + {8A0A0FBB-2B12-49B3-951E-5402D64075DD}.Release|Any CPU.Build.0 = Release|Any CPU + {FDC382B3-7255-4D33-AAA9-1168B7D01E94}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {FDC382B3-7255-4D33-AAA9-1168B7D01E94}.Debug|Any CPU.Build.0 = Debug|Any CPU + {FDC382B3-7255-4D33-AAA9-1168B7D01E94}.Release|Any CPU.ActiveCfg = Release|Any CPU + {FDC382B3-7255-4D33-AAA9-1168B7D01E94}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/README.md b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/README.md index 05117c0ea05ac..b9aab8474bd23 100644 --- a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/README.md +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/README.md @@ -1,6 +1,6 @@ # Azure Schema Registry Apache Avro client library for .NET -Azure Schema Registry is a schema repository service hosted by Azure Event Hubs, providing schema storage, versioning, and management. This package provides an Avro serializer capable of serializing and deserializing payloads containing Schema Registry schema identifiers and Avro-encoded data. +Azure Schema Registry is a schema repository service hosted by Azure Event Hubs, providing schema storage, versioning, and management. This package provides an Avro encoder capable of encoding and decoding payloads containing Schema Registry schema identifiers and Avro-encoded data. ## Getting started @@ -50,23 +50,17 @@ var schemaRegistryClient = new SchemaRegistryClient(fullyQualifiedNamespace: ful ## Key concepts -### ObjectSerializer +### Encoder -This library provides a serializer, [SchemaRegistryAvroObjectSerializer][schema_registry_avro_serializer], that implements the [ObjectSerializer][object_serializer] abstract class. This allows a developer to use this serializer in any .NET Azure SDKs that utilize ObjectSerializer. The SchemaRegistryAvroObjectSerializer utilitizes a SchemaRegistryClient to construct messages using a wire format containing schema information such as a schema ID. +This library provides an encoder, [SchemaRegistryAvroEncoder] +[schema_registry_avro_encoder], that interacts with `EventData` events. The SchemaRegistryAvroEncoder utilizes a SchemaRegistryClient to enrich the `EventData` events with the schema ID for the schema used to encode the data. -This serializer requires the [Apache Avro library][apache_avro_library]. The payload types accepted by this serializer include [GenericRecord][generic_record] and [ISpecificRecord][specific_record]. +This encoder requires the [Apache Avro library][apache_avro_library]. The payload types accepted by this encoder include [GenericRecord][generic_record] and [ISpecificRecord][specific_record]. -### Wire Format -The serializer in this library creates messages in a wire format. The format is the following: +### Examples -- Bytes [0-3] – record format indicator – currently is \x00\x00\x00\x00 -- Bytes [4-35] – UTF-8 GUID, identifying the schema in a Schema Registry instance -- Bytes [36-end] – serialized payload bytes - -## Examples - -The following shows examples of what is available through the SchemaRegistryAvroObjectSerializer. There are both sync and async methods available for these operations. These examples use a generated Apache Avro class [Employee.cs][employee] created using this schema: +The following shows examples of what is available through the `SchemaRegistryAvroEncoder`. There are both sync and async methods available for these operations. These examples use a generated Apache Avro class [Employee.cs][employee] created using this schema: ```json { @@ -82,30 +76,8 @@ The following shows examples of what is available through the SchemaRegistryAvro Details on generating a class using the Apache Avro library can be found in the [Avro C# Documentation][avro_csharp_documentation]. -* [Serialize](#serialize) -* [Deserialize](#deserialize) - -### Serialize - -Register a schema to be stored in the Azure Schema Registry. - -```C# Snippet:SchemaRegistryAvroSerialize -var employee = new Employee { Age = 42, Name = "John Doe" }; +### Encode and decode data using the Event Hub EventData model -using var memoryStream = new MemoryStream(); -var serializer = new SchemaRegistryAvroObjectSerializer(schemaRegistryClient, groupName, new SchemaRegistryAvroObjectSerializerOptions { AutoRegisterSchemas = true }); -serializer.Serialize(memoryStream, employee, typeof(Employee), CancellationToken.None); -``` - -### Deserialize - -Retrieve a previously registered schema ID from the Azure Schema Registry. - -```C# Snippet:SchemaRegistryAvroDeserialize -var serializer = new SchemaRegistryAvroObjectSerializer(schemaRegistryClient, groupName, new SchemaRegistryAvroObjectSerializerOptions { AutoRegisterSchemas = true }); -memoryStream.Position = 0; -Employee employee = (Employee)serializer.Deserialize(memoryStream, typeof(Employee), CancellationToken.None); -``` ## Troubleshooting @@ -139,8 +111,7 @@ This project has adopted the [Microsoft Open Source Code of Conduct][code_of_con [code_of_conduct]: https://opensource.microsoft.com/codeofconduct/ [code_of_conduct_faq]: https://opensource.microsoft.com/codeofconduct/faq/ [email_opencode]: mailto:opencode@microsoft.com -[object_serializer]: https://github.com/Azure/azure-sdk-for-net/blob/main/sdk/core/Azure.Core/src/Serialization/ObjectSerializer.cs -[schema_registry_avro_serializer]: https://github.com/Azure/azure-sdk-for-net/blob/main/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroObjectSerializer.cs +[schema_registry_avro_encoder]: https://github.com/Azure/azure-sdk-for-net/blob/main/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroEncoder.cs [employee]: https://github.com/Azure/azure-sdk-for-net/blob/main/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/Models/Employee.cs [avro_csharp_documentation]: https://avro.apache.org/docs/current/api/csharp/html/index.html [apache_avro_library]: https://www.nuget.org/packages/Apache.Avro/ diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/api/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.netstandard2.0.cs b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/api/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.netstandard2.0.cs index 338fd2b7eaeb8..a743cdd5ecf79 100644 --- a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/api/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.netstandard2.0.cs +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/api/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.netstandard2.0.cs @@ -1,16 +1,16 @@ namespace Microsoft.Azure.Data.SchemaRegistry.ApacheAvro { - public partial class SchemaRegistryAvroObjectSerializer : Azure.Core.Serialization.ObjectSerializer + public partial class SchemaRegistryAvroEncoder { - public SchemaRegistryAvroObjectSerializer(Azure.Data.SchemaRegistry.SchemaRegistryClient client, string groupName, Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.SchemaRegistryAvroObjectSerializerOptions options = null) { } - public override object Deserialize(System.IO.Stream stream, System.Type returnType, System.Threading.CancellationToken cancellationToken) { throw null; } - public override System.Threading.Tasks.ValueTask DeserializeAsync(System.IO.Stream stream, System.Type returnType, System.Threading.CancellationToken cancellationToken) { throw null; } - public override void Serialize(System.IO.Stream stream, object value, System.Type inputType, System.Threading.CancellationToken cancellationToken) { } - public override System.Threading.Tasks.ValueTask SerializeAsync(System.IO.Stream stream, object value, System.Type inputType, System.Threading.CancellationToken cancellationToken) { throw null; } + public SchemaRegistryAvroEncoder(Azure.Data.SchemaRegistry.SchemaRegistryClient client, string groupName, Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.SchemaRegistryAvroObjectEncoderOptions options = null) { } + public object DecodeMessageData(Azure.Messaging.IMessageWithContentType message, System.Type returnType, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; } + public System.Threading.Tasks.ValueTask DecodeMessageDataAsync(Azure.Messaging.IMessageWithContentType message, System.Type returnType, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; } + public void EncodeMessageData(Azure.Messaging.IMessageWithContentType message, object data, System.Type inputType = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { } + public System.Threading.Tasks.ValueTask EncodeMessageDataAsync(Azure.Messaging.IMessageWithContentType message, object data, System.Type inputType = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; } } - public partial class SchemaRegistryAvroObjectSerializerOptions + public partial class SchemaRegistryAvroObjectEncoderOptions { - public SchemaRegistryAvroObjectSerializerOptions() { } + public SchemaRegistryAvroObjectEncoderOptions() { } public bool AutoRegisterSchemas { get { throw null; } set { } } } } diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.csproj b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.csproj index 07990885b82d4..d37785f371c07 100644 --- a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.csproj +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.csproj @@ -8,9 +8,7 @@ - - @@ -22,5 +20,9 @@ + + + + diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/Properties/AssemblyInfo.cs b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/Properties/AssemblyInfo.cs new file mode 100644 index 0000000000000..c77ed5dbcbadf --- /dev/null +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/Properties/AssemblyInfo.cs @@ -0,0 +1,6 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.Tests, PublicKey=0024000004800000940000000602000000240000525341310004000001000100d15ddcb29688295338af4b7686603fe614abd555e09efba8fb88ee09e1f7b1ccaeed2e8f823fa9eef3fdd60217fc012ea67d2479751a0b8c087a4185541b851bd8b16f8d91b840e51b1cb0ba6fe647997e57429265e85ef62d565db50a69ae1647d54d7bd855e4db3d8a91510e5bcbd0edfbbecaa20a7bd9ae74593daa7b11b4")] diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroEncoder.cs b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroEncoder.cs new file mode 100644 index 0000000000000..3c258951a2a53 --- /dev/null +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroEncoder.cs @@ -0,0 +1,331 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +using Avro; +using Avro.Generic; +using Avro.IO; +using Avro.Specific; +using Azure.Core; +using Azure.Core.Serialization; +using Azure.Data.SchemaRegistry; +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Runtime.Serialization; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using Azure.Core.Pipeline; +using Azure.Messaging; + +namespace Microsoft.Azure.Data.SchemaRegistry.ApacheAvro +{ + /// + /// A uses the to + /// encode and decode Avro payloads. + /// + public class SchemaRegistryAvroEncoder + { + private readonly SchemaRegistryClient _client; + private readonly string _groupName; + private readonly SchemaRegistryAvroObjectEncoderOptions _options; + private const string AvroMimeType = "avro/binary"; + + /// + /// Initializes new instance of . + /// + public SchemaRegistryAvroEncoder(SchemaRegistryClient client, string groupName, SchemaRegistryAvroObjectEncoderOptions options = null) + { + _client = client ?? throw new ArgumentNullException(nameof(client)); + _groupName = groupName ?? throw new ArgumentNullException(nameof(groupName)); + _options = options; + } + + // TODO support backcompat for first beta + // private static readonly byte[] EmptyRecordFormatIndicator = { 0, 0, 0, 0 }; + // private const int RecordFormatIndicatorLength = 4; + // private const int SchemaIdLength = 32; + // private const int PayloadStartPosition = RecordFormatIndicatorLength + SchemaIdLength; + private readonly ConcurrentDictionary _idToSchemaMap = new(); + private readonly ConcurrentDictionary _schemaToIdMap = new(); + + private enum SupportedType + { + SpecificRecord, + GenericRecord + } + + private static SupportedType GetSupportedTypeOrThrow(Type type) + { + if (typeof(ISpecificRecord).IsAssignableFrom(type)) + { + return SupportedType.SpecificRecord; + } + + if (typeof(GenericRecord).IsAssignableFrom(type)) + { + return SupportedType.GenericRecord; + } + + throw new ArgumentException($"Type {type.Name} is not supported for serialization operations."); + } + + private async Task GetSchemaIdAsync(Schema schema, bool async, CancellationToken cancellationToken) + { + if (_schemaToIdMap.TryGetValue(schema, out string schemaId)) + { + return schemaId; + } + + SchemaProperties schemaProperties; + if (async) + { + schemaProperties = _options.AutoRegisterSchemas + ? (await _client + .RegisterSchemaAsync(_groupName, schema.Fullname, schema.ToString(), SchemaFormat.Avro, cancellationToken) + .ConfigureAwait(false)).Value + : await _client + .GetSchemaPropertiesAsync(_groupName, schema.Fullname, schema.ToString(), SchemaFormat.Avro, cancellationToken) + .ConfigureAwait(false); + } + else + { + schemaProperties = _options.AutoRegisterSchemas + ? _client.RegisterSchema(_groupName, schema.Fullname, schema.ToString(), SchemaFormat.Avro, cancellationToken) + : _client.GetSchemaProperties(_groupName, schema.Fullname, schema.ToString(), SchemaFormat.Avro, cancellationToken); + } + + string id = schemaProperties.Id; + + _schemaToIdMap.TryAdd(schema, id); + _idToSchemaMap.TryAdd(id, schema); + return id; + } + + private static DatumWriter GetWriterAndSchema(object value, SupportedType supportedType, out Schema schema) + { + switch (supportedType) + { + case SupportedType.SpecificRecord: + schema = ((ISpecificRecord)value).Schema; + return new SpecificDatumWriter(schema); + case SupportedType.GenericRecord: + schema = ((GenericRecord)value).Schema; + return new GenericDatumWriter(schema); + default: + throw new ArgumentException($"Invalid supported type value: {supportedType}"); + } + } + + internal (string SchemaId, BinaryData Data) Encode(object value, Type inputType, CancellationToken cancellationToken) => + EncodeInternalAsync(value, inputType, false, cancellationToken).EnsureCompleted(); + + internal async ValueTask<(string SchemaId, BinaryData Data)> EncodeAsync(object value, Type inputType, CancellationToken cancellationToken) => + await EncodeInternalAsync(value, inputType, true, cancellationToken).ConfigureAwait(false); + + private async ValueTask<(string SchemaId, BinaryData Data)> EncodeInternalAsync( + object value, + Type inputType, + bool async, + CancellationToken cancellationToken) + { + Argument.AssertNotNull(value, nameof(value)); + Argument.AssertNotNull(inputType, nameof(inputType)); + + var supportedType = GetSupportedTypeOrThrow(inputType); + var writer = GetWriterAndSchema(value, supportedType, out var schema); + + using Stream stream = new MemoryStream(); + var binaryEncoder = new BinaryEncoder(stream); + + writer.Write(value, binaryEncoder); + binaryEncoder.Flush(); + stream.Position = 0; + BinaryData data = BinaryData.FromStream(stream); + + if (async) + { + return (await GetSchemaIdAsync(schema, true, cancellationToken).ConfigureAwait(false), data); + } + else + { + return (GetSchemaIdAsync(schema, false, cancellationToken).EnsureCompleted(), data); + } + } + + private async Task GetSchemaByIdAsync(string schemaId, bool async, CancellationToken cancellationToken) + { + if (_idToSchemaMap.TryGetValue(schemaId, out Schema cachedSchema)) + { + return cachedSchema; + } + + string schemaDefinition; + if (async) + { + schemaDefinition = (await _client.GetSchemaAsync(schemaId, cancellationToken).ConfigureAwait(false)).Value.Definition; + } + else + { + schemaDefinition = _client.GetSchema(schemaId, cancellationToken).Value.Definition; + } + var schema = Schema.Parse(schemaDefinition); + _idToSchemaMap.TryAdd(schemaId, schema); + _schemaToIdMap.TryAdd(schema, schemaId); + return schema; + } + + private static DatumReader GetReader(Schema schema, SupportedType supportedType) + { + switch (supportedType) + { + case SupportedType.SpecificRecord: + return new SpecificDatumReader(schema, schema); + case SupportedType.GenericRecord: + return new GenericDatumReader(schema, schema); + default: + throw new ArgumentException($"Invalid supported type value: {supportedType}"); + } + } + + /// + /// Encodes the message data into Avro and stores it in . The + /// will be set to "avro/binary+schemaId" where schemaId is the ID of the schema used to encode the data. + /// + /// The data to serialize to Avro and encode into the message. + /// The type to use to serialize the data. + /// + /// An optional instance to signal the request to cancel the operation. + public T EncodeMessageData( + object data, + Type inputType = default, + Func func = default, + CancellationToken cancellationToken = default) where T:IMessageWithContentType + { + (string schemaId, BinaryData bd) = Encode(data, inputType ?? data?.GetType(), cancellationToken); + // IMessageWithContentType message = (IMessageWithContentType) Activator.CreateInstance(typeof(T), bd); + IMessageWithContentType message = (IMessageWithContentType) func.Invoke(bd); + + message.ContentType = $"{AvroMimeType}+{schemaId}"; + message.Data = bd; + return (T) message; + } + + /// + /// Encodes the message data into Avro and stores it in . The + /// will be set to "avro/binary+schemaId" where schemaId is the ID of the schema used to encode the data. + /// + /// The data to serialize to Avro and encode into the message. + /// The type to use to serialize the data. + /// + /// An optional instance to signal the request to cancel the operation. + public async ValueTask EncodeMessageDataAsync( + object data, + Type inputType = default, + Func func = default, + CancellationToken cancellationToken = default) + { + (string schemaId, BinaryData bd) = await EncodeAsync(data, inputType ?? data?.GetType(), cancellationToken).ConfigureAwait(false); + // IMessageWithContentType message = (IMessageWithContentType) Activator.CreateInstance(typeof(T), bd); + + IMessageWithContentType message = (IMessageWithContentType) func.Invoke(bd); + message.ContentType = $"{AvroMimeType}+{schemaId}"; + message.Data = bd; + return (T) message; + } + + /// + /// Decodes the message data into the specified type using the schema information populated in . + /// + /// The message containing the data to decode. + /// The type to deserialize to. + /// An optional instance to signal the request to cancel the operation. + /// The deserialized data. + /// Thrown if the content type is not in the expected format. + /// Thrown if an attempt is made to decode non-Avro data. + public object DecodeMessageData( + IMessageWithContentType message, + Type returnType, + CancellationToken cancellationToken = default) + => DecodeMessageBodyInternalAsync(message.Data, message.ContentType, returnType, false, cancellationToken).EnsureCompleted(); + + /// + /// Decodes the message data into the specified type using the schema information populated in . + /// + /// The message containing the data to decode. + /// The type to deserialize to. + /// An optional instance to signal the request to cancel the operation. + /// The deserialized data. + /// Thrown if the content type is not in the expected format. + /// Thrown if an attempt is made to decode non-Avro data. + public async ValueTask DecodeMessageDataAsync( + IMessageWithContentType message, + Type returnType, + CancellationToken cancellationToken = default) + => await DecodeMessageBodyInternalAsync(message.Data, message.ContentType, returnType, true, cancellationToken).ConfigureAwait(false); + + private async ValueTask DecodeMessageBodyInternalAsync( + BinaryData data, + string contentType, + Type returnType, + bool async, + CancellationToken cancellationToken) + { + string[] contentTypeArray = contentType.Split('+'); + if (contentTypeArray.Length != 2) + { + throw new FormatException("Content type was not in the expected format of MIME type + schema ID"); + } + + if (contentTypeArray[0] != AvroMimeType) + { + throw new InvalidOperationException("An avro encoder may only be used on content that is of 'avro/binary' type"); + } + + if (async) + { + return await DecodeAsync(data, contentTypeArray[1], returnType, cancellationToken).ConfigureAwait(false); + } + else + { + return Decode(data, contentTypeArray[1], returnType, cancellationToken); + } + } + + internal object Decode(BinaryData data, string schemaId, Type returnType, CancellationToken cancellationToken) => + DecodeInternalAsync(data, schemaId, returnType, false, cancellationToken).EnsureCompleted(); + + internal async ValueTask DecodeAsync(BinaryData data, string schemaId, Type returnType, CancellationToken cancellationToken) => + await DecodeInternalAsync(data, schemaId, returnType, true, cancellationToken).ConfigureAwait(false); + + private async ValueTask DecodeInternalAsync( + BinaryData data, + string schemaId, + Type returnType, + bool async, + CancellationToken cancellationToken) + { + Argument.AssertNotNull(data, nameof(data)); + Argument.AssertNotNull(returnType, nameof(returnType)); + Argument.AssertNotNull(schemaId, nameof(schemaId)); + + SupportedType supportedType = GetSupportedTypeOrThrow(returnType); + + Schema schema; + if (async) + { + schema = await GetSchemaByIdAsync(schemaId, true, cancellationToken).ConfigureAwait(false); + } + else + { + schema = GetSchemaByIdAsync(schemaId, false, cancellationToken).EnsureCompleted(); + } + + var binaryDecoder = new BinaryDecoder(data.ToStream()); + DatumReader reader = GetReader(schema, supportedType); + return reader.Read(reuse: null, binaryDecoder); + } + } +} diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroObjectSerializerOptions.cs b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroObjectEncoderOptions.cs similarity index 82% rename from sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroObjectSerializerOptions.cs rename to sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroObjectEncoderOptions.cs index 6942dcaadcbe7..f7f8475b2c9bb 100644 --- a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroObjectSerializerOptions.cs +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroObjectEncoderOptions.cs @@ -4,9 +4,9 @@ namespace Microsoft.Azure.Data.SchemaRegistry.ApacheAvro { /// - /// Options for . + /// Options for . /// - public class SchemaRegistryAvroObjectSerializerOptions + public class SchemaRegistryAvroObjectEncoderOptions { /// /// Gets or sets the automatic registration of schemas flag. diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroObjectSerializer.cs b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroObjectSerializer.cs deleted file mode 100644 index c1eefdb2a1d0c..0000000000000 --- a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroObjectSerializer.cs +++ /dev/null @@ -1,261 +0,0 @@ -// Copyright (c) Microsoft Corporation. All rights reserved. -// Licensed under the MIT License. - -using Avro; -using Avro.Generic; -using Avro.IO; -using Avro.Specific; -using Azure.Core; -using Azure.Core.Serialization; -using Azure.Data.SchemaRegistry; -using System; -using System.Collections.Concurrent; -using System.Collections.Generic; -using System.IO; -using System.Linq; -using System.Runtime.Serialization; -using System.Text; -using System.Threading; -using System.Threading.Tasks; -using Azure.Core.Pipeline; - -namespace Microsoft.Azure.Data.SchemaRegistry.ApacheAvro -{ - /// - /// A implementation that uses for Avro serialization/deserialization. - /// - public class SchemaRegistryAvroObjectSerializer : ObjectSerializer - { - private readonly SchemaRegistryClient _client; - private readonly string _groupName; - private readonly SchemaRegistryAvroObjectSerializerOptions _options; - - /// - /// Initializes new instance of . - /// - public SchemaRegistryAvroObjectSerializer(SchemaRegistryClient client, string groupName, SchemaRegistryAvroObjectSerializerOptions options = null) - { - _client = client ?? throw new ArgumentNullException(nameof(client)); - _groupName = groupName ?? throw new ArgumentNullException(nameof(groupName)); - _options = options; - } - - private static readonly byte[] EmptyRecordFormatIndicator = { 0, 0, 0, 0 }; - private static readonly Encoding Utf8Encoding = new UTF8Encoding(false); - - private const int RecordFormatIndicatorLength = 4; - private const int SchemaIdLength = 32; - private const int PayloadStartPosition = RecordFormatIndicatorLength + SchemaIdLength; - private readonly ConcurrentDictionary _idToSchemaMap = new(); - private readonly ConcurrentDictionary _schemaToIdMap = new(); - - private enum SupportedType - { - SpecificRecord, - GenericRecord - } - - private static SupportedType GetSupportedTypeOrThrow(Type type) - { - if (typeof(ISpecificRecord).IsAssignableFrom(type)) - { - return SupportedType.SpecificRecord; - } - - if (typeof(GenericRecord).IsAssignableFrom(type)) - { - return SupportedType.GenericRecord; - } - - throw new ArgumentException($"Type {type.Name} is not supported for serialization operations."); - } - - private async Task GetSchemaIdAsync(Schema schema, bool async, CancellationToken cancellationToken) - { - if (_schemaToIdMap.TryGetValue(schema, out string schemaId)) - { - return schemaId; - } - - SchemaProperties schemaProperties; - if (async) - { - schemaProperties = _options.AutoRegisterSchemas - ? (await _client - .RegisterSchemaAsync(_groupName, schema.Fullname, schema.ToString(), SchemaFormat.Avro, cancellationToken) - .ConfigureAwait(false)).Value - : await _client - .GetSchemaPropertiesAsync(_groupName, schema.Fullname, schema.ToString(), SchemaFormat.Avro, cancellationToken) - .ConfigureAwait(false); - } - else - { - schemaProperties = _options.AutoRegisterSchemas - ? _client.RegisterSchema(_groupName, schema.Fullname, schema.ToString(), SchemaFormat.Avro, cancellationToken) - : _client.GetSchemaProperties(_groupName, schema.Fullname, schema.ToString(), SchemaFormat.Avro, cancellationToken); - } - - string id = schemaProperties.Id; - - _schemaToIdMap.TryAdd(schema, id); - _idToSchemaMap.TryAdd(id, schema); - return id; - } - - private static DatumWriter GetWriterAndSchema(object value, SupportedType supportedType, out Schema schema) - { - switch (supportedType) - { - case SupportedType.SpecificRecord: - schema = ((ISpecificRecord)value).Schema; - return new SpecificDatumWriter(schema); - case SupportedType.GenericRecord: - schema = ((GenericRecord)value).Schema; - return new GenericDatumWriter(schema); - default: - throw new ArgumentException($"Invalid supported type value: {supportedType}"); - } - } - - /// - public override void Serialize(Stream stream, object value, Type inputType, CancellationToken cancellationToken) => - SerializeInternalAsync(stream, value, inputType, false, cancellationToken).EnsureCompleted(); - - /// - public override async ValueTask SerializeAsync(Stream stream, object value, Type inputType, CancellationToken cancellationToken) => - await SerializeInternalAsync(stream, value, inputType, true, cancellationToken).ConfigureAwait(false); - - private async ValueTask SerializeInternalAsync( - Stream stream, - object value, - Type inputType, - bool async, - CancellationToken cancellationToken) - { - Argument.AssertNotNull(stream, nameof(stream)); - Argument.AssertNotNull(value, nameof(value)); - Argument.AssertNotNull(inputType, nameof(inputType)); - - var supportedType = GetSupportedTypeOrThrow(inputType); - var writer = GetWriterAndSchema(value, supportedType, out var schema); - - string schemaId; - if (async) - { - schemaId = await GetSchemaIdAsync(schema, true, cancellationToken).ConfigureAwait(false); - } - else - { - schemaId = GetSchemaIdAsync(schema, false, cancellationToken).EnsureCompleted(); - } - - var binaryEncoder = new BinaryEncoder(stream); - - if (async) - { - await stream.WriteAsync(EmptyRecordFormatIndicator, 0, RecordFormatIndicatorLength, cancellationToken).ConfigureAwait(false); - await stream.WriteAsync(Utf8Encoding.GetBytes(schemaId), 0, SchemaIdLength, cancellationToken).ConfigureAwait(false); - } - else - { - stream.Write(EmptyRecordFormatIndicator, 0, RecordFormatIndicatorLength); - stream.Write(Utf8Encoding.GetBytes(schemaId), 0, SchemaIdLength); - } - - writer.Write(value, binaryEncoder); - binaryEncoder.Flush(); - } - - private async Task GetSchemaByIdAsync(string schemaId, bool async, CancellationToken cancellationToken) - { - if (_idToSchemaMap.TryGetValue(schemaId, out Schema cachedSchema)) - { - return cachedSchema; - } - - string schemaContent; - if (async) - { - schemaContent = (await _client.GetSchemaAsync(schemaId, cancellationToken).ConfigureAwait(false)).Value.Definition; - } - else - { - schemaContent = _client.GetSchema(schemaId, cancellationToken).Value.Definition; - } - var schema = Schema.Parse(schemaContent); - _idToSchemaMap.TryAdd(schemaId, schema); - _schemaToIdMap.TryAdd(schema, schemaId); - return schema; - } - - private static DatumReader GetReader(Schema schema, SupportedType supportedType) - { - switch (supportedType) - { - case SupportedType.SpecificRecord: - return new SpecificDatumReader(schema, schema); - case SupportedType.GenericRecord: - return new GenericDatumReader(schema, schema); - default: - throw new ArgumentException($"Invalid supported type value: {supportedType}"); - } - } - - private static ReadOnlyMemory CopyToReadOnlyMemory(Stream stream) - { - using var tempMemoryStream = new MemoryStream(); - stream.CopyTo(tempMemoryStream); - return new ReadOnlyMemory(tempMemoryStream.ToArray()); - } - - private static void ValidateRecordFormatIdentifier(ReadOnlyMemory message) - { - var recordFormatIdentifier = message.Slice(0, RecordFormatIndicatorLength).ToArray(); - if (!recordFormatIdentifier.SequenceEqual(EmptyRecordFormatIndicator)) - { - throw new InvalidDataContractException( - $"The record format identifier ({recordFormatIdentifier[0]:X} {recordFormatIdentifier[1]:X} {recordFormatIdentifier[2]:X} {recordFormatIdentifier[3]:X}) for the message is invalid."); - } - } - - /// - public override object Deserialize(Stream stream, Type returnType, CancellationToken cancellationToken) => - DeserializeInternalAsync(stream, returnType, false, cancellationToken).EnsureCompleted(); - - /// - public override async ValueTask DeserializeAsync(Stream stream, Type returnType, CancellationToken cancellationToken) => - await DeserializeInternalAsync(stream, returnType, true, cancellationToken).ConfigureAwait(false); - - private async ValueTask DeserializeInternalAsync( - Stream stream, - Type returnType, - bool async, - CancellationToken cancellationToken) - { - Argument.AssertNotNull(stream, nameof(stream)); - Argument.AssertNotNull(returnType, nameof(returnType)); - - SupportedType supportedType = GetSupportedTypeOrThrow(returnType); - ReadOnlyMemory message = CopyToReadOnlyMemory(stream); - ValidateRecordFormatIdentifier(message); - byte[] schemaIdBytes = message.Slice(RecordFormatIndicatorLength, SchemaIdLength).ToArray(); - string schemaId = Utf8Encoding.GetString(schemaIdBytes); - - Schema schema; - if (async) - { - schema = await GetSchemaByIdAsync(schemaId, true, cancellationToken).ConfigureAwait(false); - } - else - { - schema = GetSchemaByIdAsync(schemaId, false, cancellationToken).EnsureCompleted(); - } - - using var valueStream = new MemoryStream(message.Slice(PayloadStartPosition, message.Length - PayloadStartPosition).ToArray()); - - var binaryDecoder = new BinaryDecoder(valueStream); - DatumReader reader = GetReader(schema, supportedType); - return reader.Read(reuse: null, binaryDecoder); - } - } -} diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.Tests.csproj b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.Tests.csproj index e2008bcd1a602..f162e0912d3a2 100644 --- a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.Tests.csproj +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.Tests.csproj @@ -20,6 +20,8 @@ + + diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/Samples/Sample01_ReadmeSnippets.cs b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/Samples/Sample01_ReadmeSnippets.cs index e2fcece238f5b..36c1b562968d8 100644 --- a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/Samples/Sample01_ReadmeSnippets.cs +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/Samples/Sample01_ReadmeSnippets.cs @@ -1,12 +1,14 @@ // Copyright (c) Microsoft Corporation. All rights reserved. // Licensed under the MIT License. +using System; using Azure.Core.TestFramework; using Azure.Data.SchemaRegistry; using Azure.Identity; using NUnit.Framework; using System.IO; using System.Threading; +using Azure.Messaging.EventHubs; using TestSchema; namespace Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.Tests.Samples @@ -16,7 +18,6 @@ public class Sample01_ReadmeSnippets : SamplesBase 0); - _memoryStreamBytes = memoryStream.ToArray(); - } - - [Test] - [Order(2)] - public void Deserialize() - { - using var memoryStream = new MemoryStream(_memoryStreamBytes); - string groupName = TestEnvironment.SchemaRegistryGroup; - - #region Snippet:SchemaRegistryAvroDeserialize - var serializer = new SchemaRegistryAvroObjectSerializer(schemaRegistryClient, groupName, new SchemaRegistryAvroObjectSerializerOptions { AutoRegisterSchemas = true }); - memoryStream.Position = 0; - Employee employee = (Employee)serializer.Deserialize(memoryStream, typeof(Employee), CancellationToken.None); - #endregion - - Assert.AreEqual(42, employee.Age); - Assert.AreEqual("John Doe", employee.Name); - } + // [Test] + // public void EncodeDecodeEventData() + // { + // string groupName = TestEnvironment.SchemaRegistryGroup; + // + // #region Snippet:SchemaRegistryAvroEncodeEventData + // var employee = new Employee { Age = 42, Name = "John Doe" }; + // var encoder = new SchemaRegistryAvroEncoder(schemaRegistryClient, groupName, new SchemaRegistryAvroObjectEncoderOptions { AutoRegisterSchemas = true }); + // var eventData = new EventData(); + // + // encoder.EncodeMessageData(eventData, employee, typeof(Employee)); + // + // // the schema Id will be included as a parameter of the content type + // Console.WriteLine(eventData.ContentType); + // + // // the serialized Avro data will be stored in the EventBody + // Console.WriteLine(eventData.EventBody); + // + // // We can also get the Employee model back out from the serialized data + // employee = (Employee) encoder.DecodeMessageData(eventData, typeof(Employee)); + // #endregion + // } } } diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SchemaRegistryAvroObjectSerializerLiveTests.cs b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SchemaRegistryAvroObjectSerializerLiveTests.cs index 062cafeaf265b..a8e3f9df0f2a8 100644 --- a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SchemaRegistryAvroObjectSerializerLiveTests.cs +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SchemaRegistryAvroObjectSerializerLiveTests.cs @@ -10,6 +10,7 @@ using System.IO; using System.Threading; using System.Threading.Tasks; +using Azure.Messaging.EventHubs; using TestSchema; namespace Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.Tests @@ -35,12 +36,10 @@ public async Task CanSerializeAndDeserialize() var groupName = TestEnvironment.SchemaRegistryGroup; var employee = new Employee { Age = 42, Name = "Caketown" }; - using var memoryStream = new MemoryStream(); - var serializer = new SchemaRegistryAvroObjectSerializer(client, groupName, new SchemaRegistryAvroObjectSerializerOptions { AutoRegisterSchemas = true }); - await serializer.SerializeAsync(memoryStream, employee, typeof(Employee), CancellationToken.None); + var encoder = new SchemaRegistryAvroEncoder(client, groupName, new SchemaRegistryAvroObjectEncoderOptions() { AutoRegisterSchemas = true }); + (string schemaId, BinaryData data) = await encoder.EncodeAsync(employee, typeof(Employee), CancellationToken.None); - memoryStream.Position = 0; - var deserializedObject = await serializer.DeserializeAsync(memoryStream, typeof(Employee), CancellationToken.None); + var deserializedObject = await encoder.DecodeAsync(data, schemaId, typeof(Employee), CancellationToken.None); var readEmployee = deserializedObject as Employee; Assert.IsNotNull(readEmployee); Assert.AreEqual("Caketown", readEmployee.Name); @@ -57,11 +56,11 @@ public async Task CanSerializeAndDeserializeGenericRecord() record.Add("Age", 42); using var memoryStream = new MemoryStream(); - var serializer = new SchemaRegistryAvroObjectSerializer(client, groupName, new SchemaRegistryAvroObjectSerializerOptions { AutoRegisterSchemas = true }); - await serializer.SerializeAsync(memoryStream, record, typeof(GenericRecord), CancellationToken.None); + var encoder = new SchemaRegistryAvroEncoder(client, groupName, new SchemaRegistryAvroObjectEncoderOptions { AutoRegisterSchemas = true }); + (string schemaId, BinaryData data) = await encoder.EncodeAsync(record, typeof(GenericRecord), CancellationToken.None); memoryStream.Position = 0; - var deserializedObject = await serializer.DeserializeAsync(memoryStream, typeof(GenericRecord), CancellationToken.None); + var deserializedObject = await encoder.DecodeAsync(data, schemaId, typeof(GenericRecord), CancellationToken.None); var readRecord = deserializedObject as GenericRecord; Assert.IsNotNull(readRecord); Assert.AreEqual("Caketown", readRecord.GetValue(0)); @@ -75,9 +74,8 @@ public async Task CannotSerializeUnsupportedType() var groupName = TestEnvironment.SchemaRegistryGroup; var timeZoneInfo = TimeZoneInfo.Utc; - using var memoryStream = new MemoryStream(); - var serializer = new SchemaRegistryAvroObjectSerializer(client, groupName, new SchemaRegistryAvroObjectSerializerOptions { AutoRegisterSchemas = true }); - Assert.ThrowsAsync(async () => await serializer.SerializeAsync(memoryStream, timeZoneInfo, typeof(TimeZoneInfo), CancellationToken.None)); + var encoder = new SchemaRegistryAvroEncoder(client, groupName, new SchemaRegistryAvroObjectEncoderOptions { AutoRegisterSchemas = true }); + Assert.ThrowsAsync(async () => await encoder.EncodeAsync(timeZoneInfo, typeof(TimeZoneInfo), CancellationToken.None)); await Task.CompletedTask; } @@ -88,9 +86,52 @@ public async Task CannotDeserializeUnsupportedType() var groupName = TestEnvironment.SchemaRegistryGroup; using var memoryStream = new MemoryStream(); - var serializer = new SchemaRegistryAvroObjectSerializer(client, groupName, new SchemaRegistryAvroObjectSerializerOptions { AutoRegisterSchemas = true }); - Assert.ThrowsAsync(async () => await serializer.DeserializeAsync(memoryStream, typeof(TimeZoneInfo), CancellationToken.None)); + var serializer = new SchemaRegistryAvroEncoder(client, groupName, new SchemaRegistryAvroObjectEncoderOptions { AutoRegisterSchemas = true }); + Assert.ThrowsAsync(async () => await serializer.DecodeAsync(new BinaryData(Array.Empty()), "fakeSchemaId", typeof(TimeZoneInfo), CancellationToken.None)); await Task.CompletedTask; } + + [RecordedTest] + public async Task CannotDeserializeWithNullSchemaId() + { + var client = CreateClient(); + var groupName = TestEnvironment.SchemaRegistryGroup; + + using var memoryStream = new MemoryStream(); + var serializer = new SchemaRegistryAvroEncoder(client, groupName, new SchemaRegistryAvroObjectEncoderOptions { AutoRegisterSchemas = true }); + Assert.ThrowsAsync(async () => await serializer.DecodeAsync(new BinaryData(Array.Empty()), null, typeof(TimeZoneInfo), CancellationToken.None)); + await Task.CompletedTask; + } + + [RecordedTest] + public async Task CanUseEncoderWithEventData() + { + var client = CreateClient(); + var groupName = TestEnvironment.SchemaRegistryGroup; + + using var memoryStream = new MemoryStream(); + var encoder = new SchemaRegistryAvroEncoder(client, groupName, new SchemaRegistryAvroObjectEncoderOptions { AutoRegisterSchemas = true }); + + var employee = new Employee { Age = 42, Name = "Caketown" }; + EventData message = await encoder.EncodeMessageDataAsync(employee, func: (BinaryData bd) => new EventData(bd)); + + string[] contentType = message.ContentType.Split('+'); + Assert.AreEqual(2, contentType.Length); + Assert.AreEqual("avro/binary", contentType[0]); + Assert.IsNotEmpty(contentType[1]); + + Employee deserialized = (Employee)await encoder.DecodeMessageDataAsync(message, typeof(Employee)); + + // decoding should not alter the message + contentType = message.ContentType.Split('+'); + Assert.AreEqual(2, contentType.Length); + Assert.AreEqual("avro/binary", contentType[0]); + Assert.IsNotEmpty(contentType[1]); + + // verify the payload was decoded correctly + Assert.IsNotNull(deserialized); + Assert.AreEqual("Caketown", deserialized.Name); + Assert.AreEqual(42, deserialized.Age); + } } } diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithEventData.json b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithEventData.json index e69de29bb2d1d..17c58283f998c 100644 --- a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithEventData.json +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithEventData.json @@ -0,0 +1,52 @@ +{ + "Entries": [ + { + "RequestUri": "https://jolovschemaregistry.servicebus.windows.net/$schemaGroups/azsdk_net_test_group/schemas/TestSchema.Employee?api-version=2021-10", + "RequestMethod": "PUT", + "RequestHeaders": { + "Accept": "application/json", + "Authorization": "Sanitized", + "Content-Length": "131", + "Content-Type": "application/json; serialization=Avro", + "User-Agent": "azsdk-net-Data.SchemaRegistry/1.0.0-alpha.20211105.1 (.NET Framework 4.8.4420.0; Microsoft Windows 10.0.22000 )", + "x-ms-client-request-id": "81973e0e587452774435f602cb03e604", + "x-ms-return-client-request-id": "true" + }, + "RequestBody": { + "type": "record", + "name": "Employee", + "namespace": "TestSchema", + "fields": [ + { + "name": "Name", + "type": "string" + }, + { + "name": "Age", + "type": "int" + } + ] + }, + "StatusCode": 204, + "ResponseHeaders": { + "Content-Length": "0", + "Date": "Sat, 06 Nov 2021 00:59:13 GMT", + "Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/azsdk_net_test_group/schemas/TestSchema.Employee/versions/1?api-version=2021-10", + "Schema-Group-Name": "azsdk_net_test_group", + "Schema-Id": "45d496c4535a44e293de07484672cbe6", + "Schema-Id-Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/$schemas/45d496c4535a44e293de07484672cbe6?api-version=2021-10", + "Schema-Name": "TestSchema.Employee", + "Schema-Version": "1", + "Schema-Versions-Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/azsdk_net_test_group/schemas/TestSchema.Employee/versions?api-version=2021-10", + "Server": "Microsoft-HTTPAPI/2.0", + "Strict-Transport-Security": "max-age=31536000" + }, + "ResponseBody": [] + } + ], + "Variables": { + "RandomSeed": "349656055", + "SCHEMAREGISTRY_ENDPOINT": "jolovschemaregistry.servicebus.windows.net", + "SCHEMAREGISTRY_GROUP": "azsdk_net_test_group" + } +} \ No newline at end of file diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithEventDataAsync.json b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithEventDataAsync.json new file mode 100644 index 0000000000000..934f9bb2b049e --- /dev/null +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithEventDataAsync.json @@ -0,0 +1,52 @@ +{ + "Entries": [ + { + "RequestUri": "https://jolovschemaregistry.servicebus.windows.net/$schemaGroups/azsdk_net_test_group/schemas/TestSchema.Employee?api-version=2021-10", + "RequestMethod": "PUT", + "RequestHeaders": { + "Accept": "application/json", + "Authorization": "Sanitized", + "Content-Length": "131", + "Content-Type": "application/json; serialization=Avro", + "User-Agent": "azsdk-net-Data.SchemaRegistry/1.0.0-alpha.20211105.1 (.NET Framework 4.8.4420.0; Microsoft Windows 10.0.22000 )", + "x-ms-client-request-id": "bd6754185f077a0b0da99407ffea0671", + "x-ms-return-client-request-id": "true" + }, + "RequestBody": { + "type": "record", + "name": "Employee", + "namespace": "TestSchema", + "fields": [ + { + "name": "Name", + "type": "string" + }, + { + "name": "Age", + "type": "int" + } + ] + }, + "StatusCode": 204, + "ResponseHeaders": { + "Content-Length": "0", + "Date": "Sat, 06 Nov 2021 00:59:15 GMT", + "Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/azsdk_net_test_group/schemas/TestSchema.Employee/versions/1?api-version=2021-10", + "Schema-Group-Name": "azsdk_net_test_group", + "Schema-Id": "45d496c4535a44e293de07484672cbe6", + "Schema-Id-Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/$schemas/45d496c4535a44e293de07484672cbe6?api-version=2021-10", + "Schema-Name": "TestSchema.Employee", + "Schema-Version": "1", + "Schema-Versions-Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/azsdk_net_test_group/schemas/TestSchema.Employee/versions?api-version=2021-10", + "Server": "Microsoft-HTTPAPI/2.0", + "Strict-Transport-Security": "max-age=31536000" + }, + "ResponseBody": [] + } + ], + "Variables": { + "RandomSeed": "533134342", + "SCHEMAREGISTRY_ENDPOINT": "jolovschemaregistry.servicebus.windows.net", + "SCHEMAREGISTRY_GROUP": "azsdk_net_test_group" + } +} \ No newline at end of file diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CannotDeserializeWithNullSchemaId.json b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CannotDeserializeWithNullSchemaId.json new file mode 100644 index 0000000000000..aa134b54a5b17 --- /dev/null +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CannotDeserializeWithNullSchemaId.json @@ -0,0 +1,8 @@ +{ + "Entries": [], + "Variables": { + "RandomSeed": "699710809", + "SCHEMAREGISTRY_ENDPOINT": "jolovschemaregistry.servicebus.windows.net", + "SCHEMAREGISTRY_GROUP": "azsdk_net_test_group" + } +} \ No newline at end of file diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CannotDeserializeWithNullSchemaIdAsync.json b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CannotDeserializeWithNullSchemaIdAsync.json new file mode 100644 index 0000000000000..738c29458ba98 --- /dev/null +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CannotDeserializeWithNullSchemaIdAsync.json @@ -0,0 +1,8 @@ +{ + "Entries": [], + "Variables": { + "RandomSeed": "41747669", + "SCHEMAREGISTRY_ENDPOINT": "jolovschemaregistry.servicebus.windows.net", + "SCHEMAREGISTRY_GROUP": "azsdk_net_test_group" + } +} \ No newline at end of file diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/tests/Administration/RulePropertiesTests.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/tests/Administration/RulePropertiesTests.cs index e434fb8dbe4e0..064290f746478 100644 --- a/sdk/servicebus/Azure.Messaging.ServiceBus/tests/Administration/RulePropertiesTests.cs +++ b/sdk/servicebus/Azure.Messaging.ServiceBus/tests/Administration/RulePropertiesTests.cs @@ -4,7 +4,7 @@ using Azure.Messaging.ServiceBus.Administration; using NUnit.Framework; -namespace Azure.Messaging.ServiceBus.Tests.Management +namespace Azure.Messaging.ServiceBus.Tests.Administration { public class RulePropertiesTests { From b983ac0d26a40126f1fd40ca0d2dfd54bcfa5346 Mon Sep 17 00:00:00 2001 From: JoshLove-msft <54595583+JoshLove-msft@users.noreply.github.com> Date: Tue, 30 Nov 2021 16:46:17 -0800 Subject: [PATCH 2/6] Change IMessageWithContentType to MessageWithMetadata --- eng/Packages.Data.props | 1 + .../Azure.Core.Experimental.netstandard2.0.cs | 9 ++ .../src/MessageWithMetadata.cs | 23 ++++ .../tests/ConsumeEventTests.cs | 2 +- .../Azure.Messaging.EventHubs.Processor.sln | 7 + ...Azure.Messaging.EventHubs.Processor.csproj | 4 + ...re.Messaging.EventHubs.Shared.Tests.csproj | 2 - .../Azure.Messaging.EventHubs.sln | 14 ++ ...zure.Messaging.EventHubs.netstandard2.0.cs | 6 +- .../src/Azure.Messaging.EventHubs.csproj | 6 +- .../src/EventData.cs | 10 +- ...t.Azure.Data.SchemaRegistry.ApacheAvro.sln | 6 + .../README.md | 21 +++ ...chemaRegistry.ApacheAvro.netstandard2.0.cs | 8 +- .../src/SchemaRegistryAvroEncoder.cs | 60 +++++---- ...ata.SchemaRegistry.ApacheAvro.Tests.csproj | 1 + ...maRegistryAvroObjectSerializerLiveTests.cs | 120 +++++++++++++++++- .../CanUseEncoderWithEventDataUsingFunc.json | 52 ++++++++ ...UseEncoderWithEventDataUsingFuncAsync.json | 52 ++++++++ .../CanUseEncoderWithServiceBusMessage.json | 55 ++++++++ ...nUseEncoderWithServiceBusMessageAsync.json | 55 ++++++++ ...EncoderWithServiceBusMessageUsingFunc.json | 55 ++++++++ ...erWithServiceBusMessageUsingFuncAsync.json | 55 ++++++++ ...ure.Messaging.ServiceBus.netstandard2.0.cs | 12 +- .../src/Azure.Messaging.ServiceBus.csproj | 9 +- .../src/Primitives/ServiceBusMessage.cs | 18 ++- .../Primitives/ServiceBusReceivedMessage.cs | 21 ++- .../Administration/RulePropertiesTests.cs | 2 +- 28 files changed, 630 insertions(+), 56 deletions(-) create mode 100644 sdk/core/Azure.Core.Experimental/src/MessageWithMetadata.cs create mode 100644 sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithEventDataUsingFunc.json create mode 100644 sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithEventDataUsingFuncAsync.json create mode 100644 sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithServiceBusMessage.json create mode 100644 sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithServiceBusMessageAsync.json create mode 100644 sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithServiceBusMessageUsingFunc.json create mode 100644 sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithServiceBusMessageUsingFuncAsync.json diff --git a/eng/Packages.Data.props b/eng/Packages.Data.props index 9d3030a5796ac..4d986f54f52c5 100644 --- a/eng/Packages.Data.props +++ b/eng/Packages.Data.props @@ -82,6 +82,7 @@ + diff --git a/sdk/core/Azure.Core.Experimental/api/Azure.Core.Experimental.netstandard2.0.cs b/sdk/core/Azure.Core.Experimental/api/Azure.Core.Experimental.netstandard2.0.cs index 8aeda96796d8a..154d458b57f2b 100644 --- a/sdk/core/Azure.Core.Experimental/api/Azure.Core.Experimental.netstandard2.0.cs +++ b/sdk/core/Azure.Core.Experimental/api/Azure.Core.Experimental.netstandard2.0.cs @@ -94,3 +94,12 @@ public partial interface IMessageWithContentType System.BinaryData Data { get; set; } } } +namespace Azure.Messaging +{ + public abstract partial class MessageWithMetadata + { + protected MessageWithMetadata() { } + public abstract string ContentType { get; set; } + public abstract System.BinaryData Data { get; set; } + } +} diff --git a/sdk/core/Azure.Core.Experimental/src/MessageWithMetadata.cs b/sdk/core/Azure.Core.Experimental/src/MessageWithMetadata.cs new file mode 100644 index 0000000000000..1d35c163b1cd2 --- /dev/null +++ b/sdk/core/Azure.Core.Experimental/src/MessageWithMetadata.cs @@ -0,0 +1,23 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +using System; + +namespace Azure.Messaging +{ + /// + /// An abstraction for a message containing a content type along with its data. + /// + public abstract class MessageWithMetadata + { + /// + /// Gets or sets the message data. + /// + public abstract BinaryData Data { get; set; } + + /// + /// Gets or sets the message content type. + /// + public abstract string ContentType { get; set; } + } +} \ No newline at end of file diff --git a/sdk/eventgrid/Azure.Messaging.EventGrid/tests/ConsumeEventTests.cs b/sdk/eventgrid/Azure.Messaging.EventGrid/tests/ConsumeEventTests.cs index 3fb01d602cbbd..b582307c48a87 100644 --- a/sdk/eventgrid/Azure.Messaging.EventGrid/tests/ConsumeEventTests.cs +++ b/sdk/eventgrid/Azure.Messaging.EventGrid/tests/ConsumeEventTests.cs @@ -447,7 +447,7 @@ public void ConsumeEventHubCaptureFileCreatedEvent() Assert.NotNull(events); Assert.True(events[0].TryGetSystemEventData(out object eventData)); - Assert.AreEqual("AzureBlockBlob", (eventData as EventHubCaptureFileCreatedEventData).Fileurl); + Assert.AreEqual("AzureBlockBlob", (eventData as EventHubCaptureFileCreatedEventData).FileType); } #endregion diff --git a/sdk/eventhub/Azure.Messaging.EventHubs.Processor/Azure.Messaging.EventHubs.Processor.sln b/sdk/eventhub/Azure.Messaging.EventHubs.Processor/Azure.Messaging.EventHubs.Processor.sln index c514d9e659bfd..3b926a0bc12af 100755 --- a/sdk/eventhub/Azure.Messaging.EventHubs.Processor/Azure.Messaging.EventHubs.Processor.sln +++ b/sdk/eventhub/Azure.Messaging.EventHubs.Processor/Azure.Messaging.EventHubs.Processor.sln @@ -13,6 +13,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "External", "External", "{79 EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Azure.Messaging.EventHubs", "..\Azure.Messaging.EventHubs\src\Azure.Messaging.EventHubs.csproj", "{87A3ED70-190D-4E6B-A568-40DF5B9F3939}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Azure.Core.Experimental", "..\..\core\Azure.Core.Experimental\src\Azure.Core.Experimental.csproj", "{0BFCCE8E-85FD-4DF7-8E5D-A6CE9ACDB175}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -35,6 +37,10 @@ Global {87A3ED70-190D-4E6B-A568-40DF5B9F3939}.Debug|Any CPU.Build.0 = Debug|Any CPU {87A3ED70-190D-4E6B-A568-40DF5B9F3939}.Release|Any CPU.ActiveCfg = Release|Any CPU {87A3ED70-190D-4E6B-A568-40DF5B9F3939}.Release|Any CPU.Build.0 = Release|Any CPU + {0BFCCE8E-85FD-4DF7-8E5D-A6CE9ACDB175}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {0BFCCE8E-85FD-4DF7-8E5D-A6CE9ACDB175}.Debug|Any CPU.Build.0 = Debug|Any CPU + {0BFCCE8E-85FD-4DF7-8E5D-A6CE9ACDB175}.Release|Any CPU.ActiveCfg = Release|Any CPU + {0BFCCE8E-85FD-4DF7-8E5D-A6CE9ACDB175}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -42,6 +48,7 @@ Global GlobalSection(NestedProjects) = preSolution {7DFF0E65-DC9A-410D-9A11-AD6A06860FE1} = {797FF941-76FD-45FD-AC17-A73DFE2BA621} {87A3ED70-190D-4E6B-A568-40DF5B9F3939} = {797FF941-76FD-45FD-AC17-A73DFE2BA621} + {0BFCCE8E-85FD-4DF7-8E5D-A6CE9ACDB175} = {797FF941-76FD-45FD-AC17-A73DFE2BA621} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {44BD3BD5-61DF-464D-8627-E00B0BC4B3A3} diff --git a/sdk/eventhub/Azure.Messaging.EventHubs.Processor/src/Azure.Messaging.EventHubs.Processor.csproj b/sdk/eventhub/Azure.Messaging.EventHubs.Processor/src/Azure.Messaging.EventHubs.Processor.csproj index a364a95ae0c33..bc6b4fd2c1189 100644 --- a/sdk/eventhub/Azure.Messaging.EventHubs.Processor/src/Azure.Messaging.EventHubs.Processor.csproj +++ b/sdk/eventhub/Azure.Messaging.EventHubs.Processor/src/Azure.Messaging.EventHubs.Processor.csproj @@ -13,7 +13,11 @@ + diff --git a/sdk/eventhub/Azure.Messaging.EventHubs/Azure.Messaging.EventHubs.sln b/sdk/eventhub/Azure.Messaging.EventHubs/Azure.Messaging.EventHubs.sln index db1b53bd1fff8..6936bc419f3b7 100755 --- a/sdk/eventhub/Azure.Messaging.EventHubs/Azure.Messaging.EventHubs.sln +++ b/sdk/eventhub/Azure.Messaging.EventHubs/Azure.Messaging.EventHubs.sln @@ -11,6 +11,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Azure.Core.TestFramework", EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "External", "External", "{2DB233D3-E757-423C-8F8D-742B0AFF4713}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Azure.Core.Experimental", "..\..\core\Azure.Core.Experimental\src\Azure.Core.Experimental.csproj", "{FB10A6B3-88DD-4296-9241-53ECDF942842}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -57,6 +59,18 @@ Global {2CFDB3D6-5CFB-428C-9C89-29DD169B5433}.Release|x64.Build.0 = Release|Any CPU {2CFDB3D6-5CFB-428C-9C89-29DD169B5433}.Release|x86.ActiveCfg = Release|Any CPU {2CFDB3D6-5CFB-428C-9C89-29DD169B5433}.Release|x86.Build.0 = Release|Any CPU + {FB10A6B3-88DD-4296-9241-53ECDF942842}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {FB10A6B3-88DD-4296-9241-53ECDF942842}.Debug|Any CPU.Build.0 = Debug|Any CPU + {FB10A6B3-88DD-4296-9241-53ECDF942842}.Debug|x64.ActiveCfg = Debug|Any CPU + {FB10A6B3-88DD-4296-9241-53ECDF942842}.Debug|x64.Build.0 = Debug|Any CPU + {FB10A6B3-88DD-4296-9241-53ECDF942842}.Debug|x86.ActiveCfg = Debug|Any CPU + {FB10A6B3-88DD-4296-9241-53ECDF942842}.Debug|x86.Build.0 = Debug|Any CPU + {FB10A6B3-88DD-4296-9241-53ECDF942842}.Release|Any CPU.ActiveCfg = Release|Any CPU + {FB10A6B3-88DD-4296-9241-53ECDF942842}.Release|Any CPU.Build.0 = Release|Any CPU + {FB10A6B3-88DD-4296-9241-53ECDF942842}.Release|x64.ActiveCfg = Release|Any CPU + {FB10A6B3-88DD-4296-9241-53ECDF942842}.Release|x64.Build.0 = Release|Any CPU + {FB10A6B3-88DD-4296-9241-53ECDF942842}.Release|x86.ActiveCfg = Release|Any CPU + {FB10A6B3-88DD-4296-9241-53ECDF942842}.Release|x86.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE diff --git a/sdk/eventhub/Azure.Messaging.EventHubs/api/Azure.Messaging.EventHubs.netstandard2.0.cs b/sdk/eventhub/Azure.Messaging.EventHubs/api/Azure.Messaging.EventHubs.netstandard2.0.cs index bd74487821a91..1e5f1751c6f1d 100644 --- a/sdk/eventhub/Azure.Messaging.EventHubs/api/Azure.Messaging.EventHubs.netstandard2.0.cs +++ b/sdk/eventhub/Azure.Messaging.EventHubs/api/Azure.Messaging.EventHubs.netstandard2.0.cs @@ -1,6 +1,6 @@ namespace Azure.Messaging.EventHubs { - public partial class EventData + public partial class EventData : Azure.Messaging.MessageWithMetadata { public EventData() { } public EventData(System.BinaryData eventBody) { } @@ -14,8 +14,10 @@ public EventData(string eventBody) { } public System.ReadOnlyMemory Body { get { throw null; } } [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] public System.IO.Stream BodyAsStream { get { throw null; } } - public string ContentType { get { throw null; } set { } } + public override string ContentType { get { throw null; } set { } } public string CorrelationId { get { throw null; } set { } } + [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] + public override System.BinaryData Data { get { throw null; } set { } } public System.DateTimeOffset EnqueuedTime { get { throw null; } } public System.BinaryData EventBody { get { throw null; } set { } } public string MessageId { get { throw null; } set { } } diff --git a/sdk/eventhub/Azure.Messaging.EventHubs/src/Azure.Messaging.EventHubs.csproj b/sdk/eventhub/Azure.Messaging.EventHubs/src/Azure.Messaging.EventHubs.csproj index a0b874090cd67..546d4ef79bc6b 100644 --- a/sdk/eventhub/Azure.Messaging.EventHubs/src/Azure.Messaging.EventHubs.csproj +++ b/sdk/eventhub/Azure.Messaging.EventHubs/src/Azure.Messaging.EventHubs.csproj @@ -14,7 +14,11 @@ - + diff --git a/sdk/eventhub/Azure.Messaging.EventHubs/src/EventData.cs b/sdk/eventhub/Azure.Messaging.EventHubs/src/EventData.cs index 867678859df3b..d0db14c9490f3 100644 --- a/sdk/eventhub/Azure.Messaging.EventHubs/src/EventData.cs +++ b/sdk/eventhub/Azure.Messaging.EventHubs/src/EventData.cs @@ -19,7 +19,7 @@ namespace Azure.Messaging.EventHubs /// An Event Hubs event, encapsulating a set of data and its associated metadata. /// /// - public class EventData : IMessageWithContentType + public class EventData : MessageWithMetadata { /// The AMQP representation of the event, allowing access to additional protocol data elements not used directly by the Event Hubs client library. private readonly AmqpAnnotatedMessage _amqpMessage; @@ -49,9 +49,11 @@ public BinaryData EventBody } /// - /// + /// Hidden property that shadows the property. This is added + /// in order to inherit from . /// - public BinaryData Data + [EditorBrowsable(EditorBrowsableState.Never)] + public override BinaryData Data { get => EventBody; set => EventBody = value; @@ -80,7 +82,7 @@ public BinaryData Data /// /// RFC2046 (MIME Types) /// - public string ContentType + public override string ContentType { get { diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.sln b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.sln index 8644d54c1f449..80a209b9980c2 100644 --- a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.sln +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.sln @@ -15,6 +15,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Azure.Core.Experimental", " EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Azure.Messaging.EventHubs", "..\..\eventhub\Azure.Messaging.EventHubs\src\Azure.Messaging.EventHubs.csproj", "{FDC382B3-7255-4D33-AAA9-1168B7D01E94}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Azure.Messaging.ServiceBus", "..\..\servicebus\Azure.Messaging.ServiceBus\src\Azure.Messaging.ServiceBus.csproj", "{69E3D417-F0A9-4C6B-B9C5-8C29E9F2A660}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -45,6 +47,10 @@ Global {FDC382B3-7255-4D33-AAA9-1168B7D01E94}.Debug|Any CPU.Build.0 = Debug|Any CPU {FDC382B3-7255-4D33-AAA9-1168B7D01E94}.Release|Any CPU.ActiveCfg = Release|Any CPU {FDC382B3-7255-4D33-AAA9-1168B7D01E94}.Release|Any CPU.Build.0 = Release|Any CPU + {69E3D417-F0A9-4C6B-B9C5-8C29E9F2A660}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {69E3D417-F0A9-4C6B-B9C5-8C29E9F2A660}.Debug|Any CPU.Build.0 = Debug|Any CPU + {69E3D417-F0A9-4C6B-B9C5-8C29E9F2A660}.Release|Any CPU.ActiveCfg = Release|Any CPU + {69E3D417-F0A9-4C6B-B9C5-8C29E9F2A660}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/README.md b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/README.md index b9aab8474bd23..920effabeed5d 100644 --- a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/README.md +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/README.md @@ -78,6 +78,27 @@ Details on generating a class using the Apache Avro library can be found in the ### Encode and decode data using the Event Hub EventData model +In order to encode an `EventData` instance with Avro information, you can do the following: +```C# Snippet:SchemaRegistryAvroEncodeEventData +var encoder = new SchemaRegistryAvroEncoder(client, groupName, new SchemaRegistryAvroObjectEncoderOptions { AutoRegisterSchemas = true }); + +var employee = new Employee { Age = 42, Name = "Caketown" }; +EventData eventData = await encoder.EncodeMessageDataAsync(employee); + +// the schema Id will be included as a parameter of the content type +Console.WriteLine(eventData.ContentType); + +// the serialized Avro data will be stored in the EventBody +Console.WriteLine(eventData.EventBody); +``` + +To decode an `EventData` event that you are consuming: +```C# Snippet:SchemaRegistryAvroDecodeEventData +Employee deserialized = (Employee)await encoder.DecodeMessageDataAsync(eventData, typeof(Employee)); +Console.WriteLine(deserialized.Age); +Console.WriteLine(deserialized.Name); +``` + ## Troubleshooting diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/api/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.netstandard2.0.cs b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/api/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.netstandard2.0.cs index a743cdd5ecf79..b540da681814a 100644 --- a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/api/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.netstandard2.0.cs +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/api/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.netstandard2.0.cs @@ -3,10 +3,10 @@ namespace Microsoft.Azure.Data.SchemaRegistry.ApacheAvro public partial class SchemaRegistryAvroEncoder { public SchemaRegistryAvroEncoder(Azure.Data.SchemaRegistry.SchemaRegistryClient client, string groupName, Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.SchemaRegistryAvroObjectEncoderOptions options = null) { } - public object DecodeMessageData(Azure.Messaging.IMessageWithContentType message, System.Type returnType, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; } - public System.Threading.Tasks.ValueTask DecodeMessageDataAsync(Azure.Messaging.IMessageWithContentType message, System.Type returnType, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; } - public void EncodeMessageData(Azure.Messaging.IMessageWithContentType message, object data, System.Type inputType = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { } - public System.Threading.Tasks.ValueTask EncodeMessageDataAsync(Azure.Messaging.IMessageWithContentType message, object data, System.Type inputType = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; } + public object DecodeMessageData(Azure.Messaging.MessageWithMetadata message, System.Type returnType, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; } + public System.Threading.Tasks.ValueTask DecodeMessageDataAsync(Azure.Messaging.MessageWithMetadata message, System.Type returnType, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; } + public System.Threading.Tasks.ValueTask EncodeMessageDataAsync(object data, System.Type inputType = null, System.Func messageFactory = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) where T : Azure.Messaging.MessageWithMetadata { throw null; } + public T EncodeMessageData(object data, System.Type inputType = null, System.Func messageFactory = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) where T : Azure.Messaging.MessageWithMetadata { throw null; } } public partial class SchemaRegistryAvroObjectEncoderOptions { diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroEncoder.cs b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroEncoder.cs index 3c258951a2a53..6e27212cff907 100644 --- a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroEncoder.cs +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/src/SchemaRegistryAvroEncoder.cs @@ -191,53 +191,63 @@ private static DatumReader GetReader(Schema schema, SupportedType suppor } /// - /// Encodes the message data into Avro and stores it in . The + /// Encodes the message data into Avro and stores it in . The /// will be set to "avro/binary+schemaId" where schemaId is the ID of the schema used to encode the data. /// /// The data to serialize to Avro and encode into the message. /// The type to use to serialize the data. - /// + /// Optional func to create a derived instance of given the serialized Avro. + /// If not specified, it is assumed that the derived type has a public constructor accepting a instance. /// An optional instance to signal the request to cancel the operation. public T EncodeMessageData( object data, Type inputType = default, - Func func = default, - CancellationToken cancellationToken = default) where T:IMessageWithContentType - { - (string schemaId, BinaryData bd) = Encode(data, inputType ?? data?.GetType(), cancellationToken); - // IMessageWithContentType message = (IMessageWithContentType) Activator.CreateInstance(typeof(T), bd); - IMessageWithContentType message = (IMessageWithContentType) func.Invoke(bd); - - message.ContentType = $"{AvroMimeType}+{schemaId}"; - message.Data = bd; - return (T) message; - } + Func messageFactory = default, + CancellationToken cancellationToken = default) where T : MessageWithMetadata + => EncodeMessageDataInternalAsync(data, inputType, messageFactory, false, cancellationToken).EnsureCompleted(); /// - /// Encodes the message data into Avro and stores it in . The + /// Encodes the message data into Avro and stores it in . The /// will be set to "avro/binary+schemaId" where schemaId is the ID of the schema used to encode the data. /// /// The data to serialize to Avro and encode into the message. /// The type to use to serialize the data. - /// + /// Optional func to create a derived instance of given the serialized Avro. + /// If not specified, it is assumed that the derived type has a public constructor accepting a instance. /// An optional instance to signal the request to cancel the operation. public async ValueTask EncodeMessageDataAsync( object data, Type inputType = default, - Func func = default, - CancellationToken cancellationToken = default) + Func messageFactory = default, + CancellationToken cancellationToken = default) where T : MessageWithMetadata + => await EncodeMessageDataInternalAsync(data, inputType, messageFactory, true, cancellationToken).ConfigureAwait(false); + + private async ValueTask EncodeMessageDataInternalAsync( + object data, + Type inputType, + Func messageFactory, + bool async, + CancellationToken cancellationToken) where T : MessageWithMetadata { - (string schemaId, BinaryData bd) = await EncodeAsync(data, inputType ?? data?.GetType(), cancellationToken).ConfigureAwait(false); - // IMessageWithContentType message = (IMessageWithContentType) Activator.CreateInstance(typeof(T), bd); + (string schemaId, BinaryData bd) = async + ? await EncodeAsync(data, inputType ?? data?.GetType(), cancellationToken).ConfigureAwait(false) + : Encode(data, inputType ?? data?.GetType(), cancellationToken); - IMessageWithContentType message = (IMessageWithContentType) func.Invoke(bd); + MessageWithMetadata message; + if (messageFactory == default) + { + message = (MessageWithMetadata)Activator.CreateInstance(typeof(T), bd); + } + else + { + message = messageFactory.Invoke(bd); + } message.ContentType = $"{AvroMimeType}+{schemaId}"; - message.Data = bd; return (T) message; } /// - /// Decodes the message data into the specified type using the schema information populated in . + /// Decodes the message data into the specified type using the schema information populated in . /// /// The message containing the data to decode. /// The type to deserialize to. @@ -246,13 +256,13 @@ public async ValueTask EncodeMessageDataAsync( /// Thrown if the content type is not in the expected format. /// Thrown if an attempt is made to decode non-Avro data. public object DecodeMessageData( - IMessageWithContentType message, + MessageWithMetadata message, Type returnType, CancellationToken cancellationToken = default) => DecodeMessageBodyInternalAsync(message.Data, message.ContentType, returnType, false, cancellationToken).EnsureCompleted(); /// - /// Decodes the message data into the specified type using the schema information populated in . + /// Decodes the message data into the specified type using the schema information populated in . /// /// The message containing the data to decode. /// The type to deserialize to. @@ -261,7 +271,7 @@ public object DecodeMessageData( /// Thrown if the content type is not in the expected format. /// Thrown if an attempt is made to decode non-Avro data. public async ValueTask DecodeMessageDataAsync( - IMessageWithContentType message, + MessageWithMetadata message, Type returnType, CancellationToken cancellationToken = default) => await DecodeMessageBodyInternalAsync(message.Data, message.ContentType, returnType, true, cancellationToken).ConfigureAwait(false); diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.Tests.csproj b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.Tests.csproj index f162e0912d3a2..82ec8c83964f5 100644 --- a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.Tests.csproj +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.Tests.csproj @@ -22,6 +22,7 @@ + diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SchemaRegistryAvroObjectSerializerLiveTests.cs b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SchemaRegistryAvroObjectSerializerLiveTests.cs index a8e3f9df0f2a8..8cbef22f06a25 100644 --- a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SchemaRegistryAvroObjectSerializerLiveTests.cs +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SchemaRegistryAvroObjectSerializerLiveTests.cs @@ -11,6 +11,7 @@ using System.Threading; using System.Threading.Tasks; using Azure.Messaging.EventHubs; +using Azure.Messaging.ServiceBus; using TestSchema; namespace Microsoft.Azure.Data.SchemaRegistry.ApacheAvro.Tests @@ -55,11 +56,9 @@ public async Task CanSerializeAndDeserializeGenericRecord() record.Add("Name", "Caketown"); record.Add("Age", 42); - using var memoryStream = new MemoryStream(); var encoder = new SchemaRegistryAvroEncoder(client, groupName, new SchemaRegistryAvroObjectEncoderOptions { AutoRegisterSchemas = true }); (string schemaId, BinaryData data) = await encoder.EncodeAsync(record, typeof(GenericRecord), CancellationToken.None); - memoryStream.Position = 0; var deserializedObject = await encoder.DecodeAsync(data, schemaId, typeof(GenericRecord), CancellationToken.None); var readRecord = deserializedObject as GenericRecord; Assert.IsNotNull(readRecord); @@ -85,7 +84,6 @@ public async Task CannotDeserializeUnsupportedType() var client = CreateClient(); var groupName = TestEnvironment.SchemaRegistryGroup; - using var memoryStream = new MemoryStream(); var serializer = new SchemaRegistryAvroEncoder(client, groupName, new SchemaRegistryAvroObjectEncoderOptions { AutoRegisterSchemas = true }); Assert.ThrowsAsync(async () => await serializer.DecodeAsync(new BinaryData(Array.Empty()), "fakeSchemaId", typeof(TimeZoneInfo), CancellationToken.None)); await Task.CompletedTask; @@ -97,7 +95,6 @@ public async Task CannotDeserializeWithNullSchemaId() var client = CreateClient(); var groupName = TestEnvironment.SchemaRegistryGroup; - using var memoryStream = new MemoryStream(); var serializer = new SchemaRegistryAvroEncoder(client, groupName, new SchemaRegistryAvroObjectEncoderOptions { AutoRegisterSchemas = true }); Assert.ThrowsAsync(async () => await serializer.DecodeAsync(new BinaryData(Array.Empty()), null, typeof(TimeZoneInfo), CancellationToken.None)); await Task.CompletedTask; @@ -109,11 +106,56 @@ public async Task CanUseEncoderWithEventData() var client = CreateClient(); var groupName = TestEnvironment.SchemaRegistryGroup; - using var memoryStream = new MemoryStream(); + #region Snippet:SchemaRegistryAvroEncodeEventData var encoder = new SchemaRegistryAvroEncoder(client, groupName, new SchemaRegistryAvroObjectEncoderOptions { AutoRegisterSchemas = true }); var employee = new Employee { Age = 42, Name = "Caketown" }; - EventData message = await encoder.EncodeMessageDataAsync(employee, func: (BinaryData bd) => new EventData(bd)); + EventData eventData = await encoder.EncodeMessageDataAsync(employee); + +#if SNIPPET + // the schema Id will be included as a parameter of the content type + Console.WriteLine(eventData.ContentType); + + // the serialized Avro data will be stored in the EventBody + Console.WriteLine(eventData.EventBody); +#endif + #endregion + + string[] contentType = eventData.ContentType.Split('+'); + Assert.AreEqual(2, contentType.Length); + Assert.AreEqual("avro/binary", contentType[0]); + Assert.IsNotEmpty(contentType[1]); + +#if SNIPPET + #region Snippet:SchemaRegistryAvroDecodeEventData + Employee deserialized = (Employee)await encoder.DecodeMessageDataAsync(eventData, typeof(Employee)); + Console.WriteLine(deserialized.Age); + Console.WriteLine(deserialized.Name); + #endregion +#endif + + // decoding should not alter the message + contentType = eventData.ContentType.Split('+'); + Assert.AreEqual(2, contentType.Length); + Assert.AreEqual("avro/binary", contentType[0]); + Assert.IsNotEmpty(contentType[1]); + + // verify the payload was decoded correctly + Assert.IsNotNull(deserialized); + Assert.AreEqual("Caketown", deserialized.Name); + Assert.AreEqual(42, deserialized.Age); + } + + [RecordedTest] + public async Task CanUseEncoderWithEventDataUsingFunc() + { + var client = CreateClient(); + var groupName = TestEnvironment.SchemaRegistryGroup; + + var encoder = new SchemaRegistryAvroEncoder(client, groupName, new SchemaRegistryAvroObjectEncoderOptions { AutoRegisterSchemas = true }); + + var employee = new Employee { Age = 42, Name = "Caketown" }; + EventData message = await encoder.EncodeMessageDataAsync(employee, messageFactory: bd => new EventData(bd)); string[] contentType = message.ContentType.Split('+'); Assert.AreEqual(2, contentType.Length); @@ -133,5 +175,71 @@ public async Task CanUseEncoderWithEventData() Assert.AreEqual("Caketown", deserialized.Name); Assert.AreEqual(42, deserialized.Age); } + + [RecordedTest] + public async Task CanUseEncoderWithServiceBusMessage() + { + var client = CreateClient(); + var groupName = TestEnvironment.SchemaRegistryGroup; + + var encoder = new SchemaRegistryAvroEncoder(client, groupName, new SchemaRegistryAvroObjectEncoderOptions { AutoRegisterSchemas = true }); + + var employee = new Employee { Age = 42, Name = "Caketown" }; + ServiceBusMessage message = await encoder.EncodeMessageDataAsync(employee); + + string[] contentType = message.ContentType.Split('+'); + Assert.AreEqual(2, contentType.Length); + Assert.AreEqual("avro/binary", contentType[0]); + Assert.IsNotEmpty(contentType[1]); + + ServiceBusReceivedMessage receivedMessage = + ServiceBusModelFactory.ServiceBusReceivedMessage(body: message.Body, contentType: message.ContentType); + + Employee deserialized = (Employee)await encoder.DecodeMessageDataAsync(receivedMessage, typeof(Employee)); + + // decoding should not alter the message + contentType = message.ContentType.Split('+'); + Assert.AreEqual(2, contentType.Length); + Assert.AreEqual("avro/binary", contentType[0]); + Assert.IsNotEmpty(contentType[1]); + + // verify the payload was decoded correctly + Assert.IsNotNull(deserialized); + Assert.AreEqual("Caketown", deserialized.Name); + Assert.AreEqual(42, deserialized.Age); + } + + [RecordedTest] + public async Task CanUseEncoderWithServiceBusMessageUsingFunc() + { + var client = CreateClient(); + var groupName = TestEnvironment.SchemaRegistryGroup; + + var encoder = new SchemaRegistryAvroEncoder(client, groupName, new SchemaRegistryAvroObjectEncoderOptions { AutoRegisterSchemas = true }); + + var employee = new Employee { Age = 42, Name = "Caketown" }; + ServiceBusMessage message = await encoder.EncodeMessageDataAsync(employee, messageFactory: bd => new ServiceBusMessage(bd)); + + string[] contentType = message.ContentType.Split('+'); + Assert.AreEqual(2, contentType.Length); + Assert.AreEqual("avro/binary", contentType[0]); + Assert.IsNotEmpty(contentType[1]); + + ServiceBusReceivedMessage receivedMessage = + ServiceBusModelFactory.ServiceBusReceivedMessage(body: message.Body, contentType: message.ContentType); + + Employee deserialized = (Employee)await encoder.DecodeMessageDataAsync(receivedMessage, typeof(Employee)); + + // decoding should not alter the message + contentType = message.ContentType.Split('+'); + Assert.AreEqual(2, contentType.Length); + Assert.AreEqual("avro/binary", contentType[0]); + Assert.IsNotEmpty(contentType[1]); + + // verify the payload was decoded correctly + Assert.IsNotNull(deserialized); + Assert.AreEqual("Caketown", deserialized.Name); + Assert.AreEqual(42, deserialized.Age); + } } } diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithEventDataUsingFunc.json b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithEventDataUsingFunc.json new file mode 100644 index 0000000000000..2f6c634071eaf --- /dev/null +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithEventDataUsingFunc.json @@ -0,0 +1,52 @@ +{ + "Entries": [ + { + "RequestUri": "https://jolovschemaregistry.servicebus.windows.net/$schemaGroups/azsdk_net_test_group/schemas/TestSchema.Employee?api-version=2021-10", + "RequestMethod": "PUT", + "RequestHeaders": { + "Accept": "application/json", + "Authorization": "Sanitized", + "Content-Length": "131", + "Content-Type": "application/json; serialization=Avro", + "User-Agent": "azsdk-net-Data.SchemaRegistry/1.1.0-alpha.20211129.1 (.NET Framework 4.8.4420.0; Microsoft Windows 10.0.22000 )", + "x-ms-client-request-id": "63f26cbe7b57753aee9f2b29a06d502a", + "x-ms-return-client-request-id": "true" + }, + "RequestBody": { + "type": "record", + "name": "Employee", + "namespace": "TestSchema", + "fields": [ + { + "name": "Name", + "type": "string" + }, + { + "name": "Age", + "type": "int" + } + ] + }, + "StatusCode": 204, + "ResponseHeaders": { + "Content-Length": "0", + "Date": "Tue, 30 Nov 2021 00:23:11 GMT", + "Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/azsdk_net_test_group/schemas/TestSchema.Employee/versions/1?api-version=2021-10", + "Schema-Group-Name": "azsdk_net_test_group", + "Schema-Id": "a2a03a64e7b6441fa09abc8891469504", + "Schema-Id-Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/$schemas/a2a03a64e7b6441fa09abc8891469504?api-version=2021-10", + "Schema-Name": "TestSchema.Employee", + "Schema-Version": "1", + "Schema-Versions-Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/azsdk_net_test_group/schemas/TestSchema.Employee/versions?api-version=2021-10", + "Server": "Microsoft-HTTPAPI/2.0", + "Strict-Transport-Security": "max-age=31536000" + }, + "ResponseBody": [] + } + ], + "Variables": { + "RandomSeed": "822817808", + "SCHEMAREGISTRY_ENDPOINT": "jolovschemaregistry.servicebus.windows.net", + "SCHEMAREGISTRY_GROUP": "azsdk_net_test_group" + } +} \ No newline at end of file diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithEventDataUsingFuncAsync.json b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithEventDataUsingFuncAsync.json new file mode 100644 index 0000000000000..80026390d3a9f --- /dev/null +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithEventDataUsingFuncAsync.json @@ -0,0 +1,52 @@ +{ + "Entries": [ + { + "RequestUri": "https://jolovschemaregistry.servicebus.windows.net/$schemaGroups/azsdk_net_test_group/schemas/TestSchema.Employee?api-version=2021-10", + "RequestMethod": "PUT", + "RequestHeaders": { + "Accept": "application/json", + "Authorization": "Sanitized", + "Content-Length": "131", + "Content-Type": "application/json; serialization=Avro", + "User-Agent": "azsdk-net-Data.SchemaRegistry/1.1.0-alpha.20211129.1 (.NET Framework 4.8.4420.0; Microsoft Windows 10.0.22000 )", + "x-ms-client-request-id": "0d15d1dcbbffe08f0834b0a18fffdcd6", + "x-ms-return-client-request-id": "true" + }, + "RequestBody": { + "type": "record", + "name": "Employee", + "namespace": "TestSchema", + "fields": [ + { + "name": "Name", + "type": "string" + }, + { + "name": "Age", + "type": "int" + } + ] + }, + "StatusCode": 204, + "ResponseHeaders": { + "Content-Length": "0", + "Date": "Tue, 30 Nov 2021 00:23:57 GMT", + "Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/azsdk_net_test_group/schemas/TestSchema.Employee/versions/1?api-version=2021-10", + "Schema-Group-Name": "azsdk_net_test_group", + "Schema-Id": "a2a03a64e7b6441fa09abc8891469504", + "Schema-Id-Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/$schemas/a2a03a64e7b6441fa09abc8891469504?api-version=2021-10", + "Schema-Name": "TestSchema.Employee", + "Schema-Version": "1", + "Schema-Versions-Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/azsdk_net_test_group/schemas/TestSchema.Employee/versions?api-version=2021-10", + "Server": "Microsoft-HTTPAPI/2.0", + "Strict-Transport-Security": "max-age=31536000" + }, + "ResponseBody": [] + } + ], + "Variables": { + "RandomSeed": "1093756092", + "SCHEMAREGISTRY_ENDPOINT": "jolovschemaregistry.servicebus.windows.net", + "SCHEMAREGISTRY_GROUP": "azsdk_net_test_group" + } +} \ No newline at end of file diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithServiceBusMessage.json b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithServiceBusMessage.json new file mode 100644 index 0000000000000..b0a09db125f33 --- /dev/null +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithServiceBusMessage.json @@ -0,0 +1,55 @@ +{ + "Entries": [ + { + "RequestUri": "https://jolovschemaregistry.servicebus.windows.net/$schemaGroups/azsdk_net_test_group/schemas/TestSchema.Employee?api-version=2021-10", + "RequestMethod": "PUT", + "RequestHeaders": { + "Accept": "application/json", + "Authorization": "Sanitized", + "Content-Length": "131", + "Content-Type": "application/json; serialization=Avro", + "User-Agent": [ + "azsdk-net-Data.SchemaRegistry/1.1.0-alpha.20211129.1", + "(.NET Core 3.1.21; Microsoft Windows 10.0.22000)" + ], + "x-ms-client-request-id": "1b2a58488352ab38675dde76f1cb5061", + "x-ms-return-client-request-id": "true" + }, + "RequestBody": { + "type": "record", + "name": "Employee", + "namespace": "TestSchema", + "fields": [ + { + "name": "Name", + "type": "string" + }, + { + "name": "Age", + "type": "int" + } + ] + }, + "StatusCode": 204, + "ResponseHeaders": { + "Content-Length": "0", + "Date": "Tue, 30 Nov 2021 01:12:57 GMT", + "Location": "https://jolovschemaregistry.servicebus.windows.net/$schemagroups/azsdk_net_test_group/schemas/TestSchema.Employee/versions/1?api-version=2021-10", + "Schema-Group-Name": "azsdk_net_test_group", + "Schema-Id": "a2a03a64e7b6441fa09abc8891469504", + "Schema-Id-Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/$schemas/a2a03a64e7b6441fa09abc8891469504?api-version=2021-10", + "Schema-Name": "TestSchema.Employee", + "Schema-Version": "1", + "Schema-Versions-Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/azsdk_net_test_group/schemas/TestSchema.Employee/versions?api-version=2021-10", + "Server": "Microsoft-HTTPAPI/2.0", + "Strict-Transport-Security": "max-age=31536000" + }, + "ResponseBody": [] + } + ], + "Variables": { + "RandomSeed": "1905931033", + "SCHEMAREGISTRY_ENDPOINT": "jolovschemaregistry.servicebus.windows.net", + "SCHEMAREGISTRY_GROUP": "azsdk_net_test_group" + } +} \ No newline at end of file diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithServiceBusMessageAsync.json b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithServiceBusMessageAsync.json new file mode 100644 index 0000000000000..9a4586b8e85fc --- /dev/null +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithServiceBusMessageAsync.json @@ -0,0 +1,55 @@ +{ + "Entries": [ + { + "RequestUri": "https://jolovschemaregistry.servicebus.windows.net/$schemaGroups/azsdk_net_test_group/schemas/TestSchema.Employee?api-version=2021-10", + "RequestMethod": "PUT", + "RequestHeaders": { + "Accept": "application/json", + "Authorization": "Sanitized", + "Content-Length": "131", + "Content-Type": "application/json; serialization=Avro", + "User-Agent": [ + "azsdk-net-Data.SchemaRegistry/1.1.0-alpha.20211129.1", + "(.NET Core 3.1.21; Microsoft Windows 10.0.22000)" + ], + "x-ms-client-request-id": "a13002d44b162dc0ab8cdaee93264afb", + "x-ms-return-client-request-id": "true" + }, + "RequestBody": { + "type": "record", + "name": "Employee", + "namespace": "TestSchema", + "fields": [ + { + "name": "Name", + "type": "string" + }, + { + "name": "Age", + "type": "int" + } + ] + }, + "StatusCode": 204, + "ResponseHeaders": { + "Content-Length": "0", + "Date": "Tue, 30 Nov 2021 01:12:58 GMT", + "Location": "https://jolovschemaregistry.servicebus.windows.net/$schemagroups/azsdk_net_test_group/schemas/TestSchema.Employee/versions/1?api-version=2021-10", + "Schema-Group-Name": "azsdk_net_test_group", + "Schema-Id": "a2a03a64e7b6441fa09abc8891469504", + "Schema-Id-Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/$schemas/a2a03a64e7b6441fa09abc8891469504?api-version=2021-10", + "Schema-Name": "TestSchema.Employee", + "Schema-Version": "1", + "Schema-Versions-Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/azsdk_net_test_group/schemas/TestSchema.Employee/versions?api-version=2021-10", + "Server": "Microsoft-HTTPAPI/2.0", + "Strict-Transport-Security": "max-age=31536000" + }, + "ResponseBody": [] + } + ], + "Variables": { + "RandomSeed": "1063570874", + "SCHEMAREGISTRY_ENDPOINT": "jolovschemaregistry.servicebus.windows.net", + "SCHEMAREGISTRY_GROUP": "azsdk_net_test_group" + } +} \ No newline at end of file diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithServiceBusMessageUsingFunc.json b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithServiceBusMessageUsingFunc.json new file mode 100644 index 0000000000000..c7d5be53555c1 --- /dev/null +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithServiceBusMessageUsingFunc.json @@ -0,0 +1,55 @@ +{ + "Entries": [ + { + "RequestUri": "https://jolovschemaregistry.servicebus.windows.net/$schemaGroups/azsdk_net_test_group/schemas/TestSchema.Employee?api-version=2021-10", + "RequestMethod": "PUT", + "RequestHeaders": { + "Accept": "application/json", + "Authorization": "Sanitized", + "Content-Length": "131", + "Content-Type": "application/json; serialization=Avro", + "User-Agent": [ + "azsdk-net-Data.SchemaRegistry/1.1.0-alpha.20211129.1", + "(.NET Core 3.1.21; Microsoft Windows 10.0.22000)" + ], + "x-ms-client-request-id": "c7798168c10dc0828a8c6bb785e5b4a1", + "x-ms-return-client-request-id": "true" + }, + "RequestBody": { + "type": "record", + "name": "Employee", + "namespace": "TestSchema", + "fields": [ + { + "name": "Name", + "type": "string" + }, + { + "name": "Age", + "type": "int" + } + ] + }, + "StatusCode": 204, + "ResponseHeaders": { + "Content-Length": "0", + "Date": "Tue, 30 Nov 2021 01:12:58 GMT", + "Location": "https://jolovschemaregistry.servicebus.windows.net/$schemagroups/azsdk_net_test_group/schemas/TestSchema.Employee/versions/1?api-version=2021-10", + "Schema-Group-Name": "azsdk_net_test_group", + "Schema-Id": "a2a03a64e7b6441fa09abc8891469504", + "Schema-Id-Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/$schemas/a2a03a64e7b6441fa09abc8891469504?api-version=2021-10", + "Schema-Name": "TestSchema.Employee", + "Schema-Version": "1", + "Schema-Versions-Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/azsdk_net_test_group/schemas/TestSchema.Employee/versions?api-version=2021-10", + "Server": "Microsoft-HTTPAPI/2.0", + "Strict-Transport-Security": "max-age=31536000" + }, + "ResponseBody": [] + } + ], + "Variables": { + "RandomSeed": "462918451", + "SCHEMAREGISTRY_ENDPOINT": "jolovschemaregistry.servicebus.windows.net", + "SCHEMAREGISTRY_GROUP": "azsdk_net_test_group" + } +} \ No newline at end of file diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithServiceBusMessageUsingFuncAsync.json b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithServiceBusMessageUsingFuncAsync.json new file mode 100644 index 0000000000000..2075b990a7ae6 --- /dev/null +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SessionRecords/SchemaRegistryAvroObjectSerializerLiveTests/CanUseEncoderWithServiceBusMessageUsingFuncAsync.json @@ -0,0 +1,55 @@ +{ + "Entries": [ + { + "RequestUri": "https://jolovschemaregistry.servicebus.windows.net/$schemaGroups/azsdk_net_test_group/schemas/TestSchema.Employee?api-version=2021-10", + "RequestMethod": "PUT", + "RequestHeaders": { + "Accept": "application/json", + "Authorization": "Sanitized", + "Content-Length": "131", + "Content-Type": "application/json; serialization=Avro", + "User-Agent": [ + "azsdk-net-Data.SchemaRegistry/1.1.0-alpha.20211129.1", + "(.NET Core 3.1.21; Microsoft Windows 10.0.22000)" + ], + "x-ms-client-request-id": "24845f4b598e9b74e5e36e9a077bc01f", + "x-ms-return-client-request-id": "true" + }, + "RequestBody": { + "type": "record", + "name": "Employee", + "namespace": "TestSchema", + "fields": [ + { + "name": "Name", + "type": "string" + }, + { + "name": "Age", + "type": "int" + } + ] + }, + "StatusCode": 204, + "ResponseHeaders": { + "Content-Length": "0", + "Date": "Tue, 30 Nov 2021 01:13:00 GMT", + "Location": "https://jolovschemaregistry.servicebus.windows.net/$schemagroups/azsdk_net_test_group/schemas/TestSchema.Employee/versions/1?api-version=2021-10", + "Schema-Group-Name": "azsdk_net_test_group", + "Schema-Id": "a2a03a64e7b6441fa09abc8891469504", + "Schema-Id-Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/$schemas/a2a03a64e7b6441fa09abc8891469504?api-version=2021-10", + "Schema-Name": "TestSchema.Employee", + "Schema-Version": "1", + "Schema-Versions-Location": "https://jolovschemaregistry.servicebus.windows.net:443/$schemagroups/azsdk_net_test_group/schemas/TestSchema.Employee/versions?api-version=2021-10", + "Server": "Microsoft-HTTPAPI/2.0", + "Strict-Transport-Security": "max-age=31536000" + }, + "ResponseBody": [] + } + ], + "Variables": { + "RandomSeed": "2078853961", + "SCHEMAREGISTRY_ENDPOINT": "jolovschemaregistry.servicebus.windows.net", + "SCHEMAREGISTRY_GROUP": "azsdk_net_test_group" + } +} \ No newline at end of file diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/api/Azure.Messaging.ServiceBus.netstandard2.0.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/api/Azure.Messaging.ServiceBus.netstandard2.0.cs index 286d05358816c..8a2251c75ca02 100644 --- a/sdk/servicebus/Azure.Messaging.ServiceBus/api/Azure.Messaging.ServiceBus.netstandard2.0.cs +++ b/sdk/servicebus/Azure.Messaging.ServiceBus/api/Azure.Messaging.ServiceBus.netstandard2.0.cs @@ -152,7 +152,7 @@ public enum ServiceBusFailureReason SessionLockLost = 11, MessagingEntityAlreadyExists = 12, } - public partial class ServiceBusMessage + public partial class ServiceBusMessage : Azure.Messaging.MessageWithMetadata { public ServiceBusMessage() { } public ServiceBusMessage(Azure.Messaging.ServiceBus.ServiceBusReceivedMessage receivedMessage) { } @@ -161,8 +161,10 @@ public ServiceBusMessage(System.ReadOnlyMemory body) { } public ServiceBusMessage(string body) { } public System.Collections.Generic.IDictionary ApplicationProperties { get { throw null; } } public System.BinaryData Body { get { throw null; } set { } } - public string ContentType { get { throw null; } set { } } + public override string ContentType { get { throw null; } set { } } public string CorrelationId { get { throw null; } set { } } + [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] + public override System.BinaryData Data { get { throw null; } set { } } public string MessageId { get { throw null; } set { } } public string PartitionKey { get { throw null; } set { } } public string ReplyTo { get { throw null; } set { } } @@ -255,13 +257,15 @@ public ServiceBusProcessorOptions() { } [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] public override string ToString() { throw null; } } - public partial class ServiceBusReceivedMessage + public partial class ServiceBusReceivedMessage : Azure.Messaging.MessageWithMetadata { internal ServiceBusReceivedMessage() { } public System.Collections.Generic.IReadOnlyDictionary ApplicationProperties { get { throw null; } } public System.BinaryData Body { get { throw null; } } - public string ContentType { get { throw null; } } + public override string ContentType { get { throw null; } [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] set { } } public string CorrelationId { get { throw null; } } + [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] + public override System.BinaryData Data { get { throw null; } set { } } public string DeadLetterErrorDescription { get { throw null; } } public string DeadLetterReason { get { throw null; } } public string DeadLetterSource { get { throw null; } } diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Azure.Messaging.ServiceBus.csproj b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Azure.Messaging.ServiceBus.csproj index e5d4defdb7309..091ab2602d1d6 100644 --- a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Azure.Messaging.ServiceBus.csproj +++ b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Azure.Messaging.ServiceBus.csproj @@ -19,7 +19,11 @@ - + @@ -57,5 +61,8 @@ Resources.Designer.cs + + + diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusMessage.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusMessage.cs index 8cb14cadabbba..6413ee5920cb9 100755 --- a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusMessage.cs +++ b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusMessage.cs @@ -3,6 +3,7 @@ using System; using System.Collections.Generic; +using System.ComponentModel; using System.Globalization; using System.Text; using Azure.Core; @@ -19,7 +20,7 @@ namespace Azure.Messaging.ServiceBus /// The message structure is discussed in detail in the /// product documentation. /// - public class ServiceBusMessage + public class ServiceBusMessage : MessageWithMetadata { /// /// Creates a new message. @@ -157,6 +158,17 @@ public BinaryData Body } } + /// + /// Hidden property that shadows the property. This is added + /// in order to inherit from . + /// + [EditorBrowsable(EditorBrowsableState.Never)] + public override BinaryData Data + { + get => Body; + set => Body = value; + } + /// /// Gets or sets the MessageId to identify the message. /// @@ -349,7 +361,7 @@ public string To /// Optionally describes the payload of the message, with a descriptor following the format of /// RFC2045, Section 5, for example "application/json". /// - public string ContentType + public override string ContentType { get { @@ -440,4 +452,4 @@ public override string ToString() return string.Format(CultureInfo.CurrentCulture, "{{MessageId:{0}}}", MessageId); } } -} +} \ No newline at end of file diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusReceivedMessage.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusReceivedMessage.cs index ec663cdf6a187..2d4ff977e8f57 100644 --- a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusReceivedMessage.cs +++ b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusReceivedMessage.cs @@ -4,6 +4,7 @@ using System; using System.Collections.Generic; using System.Collections.ObjectModel; +using System.ComponentModel; using System.Globalization; using Azure.Core; using Azure.Core.Amqp; @@ -19,7 +20,7 @@ namespace Azure.Messaging.ServiceBus /// The message structure is discussed in detail in the /// product documentation. /// - public class ServiceBusReceivedMessage + public class ServiceBusReceivedMessage : MessageWithMetadata { /// /// Creates a new message from the specified payload. @@ -67,6 +68,17 @@ internal ServiceBusReceivedMessage(): this(body: default) /// public BinaryData Body => AmqpMessage.GetBody(); + /// + /// Hidden property that shadows the property. This is added + /// in order to inherit from . + /// + [EditorBrowsable(EditorBrowsableState.Never)] + public override BinaryData Data + { + get => Body; + set => throw new NotImplementedException("Data cannot be set on a ServiceBusReceivedMessage"); + } + /// /// Gets the MessageId to identify the message. /// @@ -166,7 +178,12 @@ internal ServiceBusReceivedMessage(): this(body: default) /// Optionally describes the payload of the message, with a descriptor following the format of /// RFC2045, Section 5, for example "application/json". /// - public string ContentType => AmqpMessage.Properties.ContentType; + public override string ContentType + { + get => AmqpMessage.Properties.ContentType; + [EditorBrowsable(EditorBrowsableState.Never)] + set => throw new NotImplementedException("Content type cannot be set on a ServiceBusReceivedMessage"); + } /// Gets the address of an entity to send replies to. /// The reply entity address. diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/tests/Administration/RulePropertiesTests.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/tests/Administration/RulePropertiesTests.cs index 064290f746478..e434fb8dbe4e0 100644 --- a/sdk/servicebus/Azure.Messaging.ServiceBus/tests/Administration/RulePropertiesTests.cs +++ b/sdk/servicebus/Azure.Messaging.ServiceBus/tests/Administration/RulePropertiesTests.cs @@ -4,7 +4,7 @@ using Azure.Messaging.ServiceBus.Administration; using NUnit.Framework; -namespace Azure.Messaging.ServiceBus.Tests.Administration +namespace Azure.Messaging.ServiceBus.Tests.Management { public class RulePropertiesTests { From fd0948c4d80ee2d6580569e361191ba7ec241f50 Mon Sep 17 00:00:00 2001 From: JoshLove-msft <54595583+JoshLove-msft@users.noreply.github.com> Date: Tue, 30 Nov 2021 17:02:17 -0800 Subject: [PATCH 3/6] Add IsReadOnly --- .../api/Azure.Core.Experimental.netstandard2.0.cs | 4 +--- .../Azure.Core.Experimental/src/MessageWithMetadata.cs | 5 +++++ .../api/Azure.Messaging.EventHubs.netstandard2.0.cs | 2 ++ sdk/eventhub/Azure.Messaging.EventHubs/src/EventData.cs | 7 +++++++ .../tests/SchemaRegistryAvroObjectSerializerLiveTests.cs | 8 ++++++-- .../Azure.Messaging.ServiceBus.sln | 6 ++++++ .../api/Azure.Messaging.ServiceBus.netstandard2.0.cs | 4 ++++ .../src/Primitives/ServiceBusMessage.cs | 7 +++++++ .../src/Primitives/ServiceBusReceivedMessage.cs | 7 +++++++ 9 files changed, 45 insertions(+), 5 deletions(-) diff --git a/sdk/core/Azure.Core.Experimental/api/Azure.Core.Experimental.netstandard2.0.cs b/sdk/core/Azure.Core.Experimental/api/Azure.Core.Experimental.netstandard2.0.cs index 154d458b57f2b..a224f37a6f6cb 100644 --- a/sdk/core/Azure.Core.Experimental/api/Azure.Core.Experimental.netstandard2.0.cs +++ b/sdk/core/Azure.Core.Experimental/api/Azure.Core.Experimental.netstandard2.0.cs @@ -93,13 +93,11 @@ public partial interface IMessageWithContentType string ContentType { get; set; } System.BinaryData Data { get; set; } } -} -namespace Azure.Messaging -{ public abstract partial class MessageWithMetadata { protected MessageWithMetadata() { } public abstract string ContentType { get; set; } public abstract System.BinaryData Data { get; set; } + public abstract bool IsReadOnly { get; } } } diff --git a/sdk/core/Azure.Core.Experimental/src/MessageWithMetadata.cs b/sdk/core/Azure.Core.Experimental/src/MessageWithMetadata.cs index 1d35c163b1cd2..6e82376136b67 100644 --- a/sdk/core/Azure.Core.Experimental/src/MessageWithMetadata.cs +++ b/sdk/core/Azure.Core.Experimental/src/MessageWithMetadata.cs @@ -19,5 +19,10 @@ public abstract class MessageWithMetadata /// Gets or sets the message content type. /// public abstract string ContentType { get; set; } + + /// + /// Gets whether the message is read only or not. + /// + public abstract bool IsReadOnly { get; } } } \ No newline at end of file diff --git a/sdk/eventhub/Azure.Messaging.EventHubs/api/Azure.Messaging.EventHubs.netstandard2.0.cs b/sdk/eventhub/Azure.Messaging.EventHubs/api/Azure.Messaging.EventHubs.netstandard2.0.cs index 1e5f1751c6f1d..66bcec52468ef 100644 --- a/sdk/eventhub/Azure.Messaging.EventHubs/api/Azure.Messaging.EventHubs.netstandard2.0.cs +++ b/sdk/eventhub/Azure.Messaging.EventHubs/api/Azure.Messaging.EventHubs.netstandard2.0.cs @@ -20,6 +20,8 @@ public EventData(string eventBody) { } public override System.BinaryData Data { get { throw null; } set { } } public System.DateTimeOffset EnqueuedTime { get { throw null; } } public System.BinaryData EventBody { get { throw null; } set { } } + [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] + public override bool IsReadOnly { get { throw null; } } public string MessageId { get { throw null; } set { } } public long Offset { get { throw null; } } public string PartitionKey { get { throw null; } } diff --git a/sdk/eventhub/Azure.Messaging.EventHubs/src/EventData.cs b/sdk/eventhub/Azure.Messaging.EventHubs/src/EventData.cs index d0db14c9490f3..78d0cf1d6ad19 100644 --- a/sdk/eventhub/Azure.Messaging.EventHubs/src/EventData.cs +++ b/sdk/eventhub/Azure.Messaging.EventHubs/src/EventData.cs @@ -107,6 +107,13 @@ public override string ContentType } } + /// + /// Hidden property that indicates that the is not read-only. This is part of + /// the abstraction. + /// + [EditorBrowsable(EditorBrowsableState.Never)] + public override bool IsReadOnly => false; + /// /// An application-defined value that uniquely identifies the event. The identifier is /// a free-form value and can reflect a GUID or an identifier derived from the application diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SchemaRegistryAvroObjectSerializerLiveTests.cs b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SchemaRegistryAvroObjectSerializerLiveTests.cs index 8cbef22f06a25..52aa5d543b032 100644 --- a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SchemaRegistryAvroObjectSerializerLiveTests.cs +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/SchemaRegistryAvroObjectSerializerLiveTests.cs @@ -10,6 +10,7 @@ using System.IO; using System.Threading; using System.Threading.Tasks; +using Azure.Messaging; using Azure.Messaging.EventHubs; using Azure.Messaging.ServiceBus; using TestSchema; @@ -121,18 +122,19 @@ public async Task CanUseEncoderWithEventData() #endif #endregion + Assert.IsFalse(((MessageWithMetadata) eventData).IsReadOnly); string[] contentType = eventData.ContentType.Split('+'); Assert.AreEqual(2, contentType.Length); Assert.AreEqual("avro/binary", contentType[0]); Assert.IsNotEmpty(contentType[1]); -#if SNIPPET #region Snippet:SchemaRegistryAvroDecodeEventData Employee deserialized = (Employee)await encoder.DecodeMessageDataAsync(eventData, typeof(Employee)); +#if SNIPPET Console.WriteLine(deserialized.Age); Console.WriteLine(deserialized.Name); - #endregion #endif + #endregion // decoding should not alter the message contentType = eventData.ContentType.Split('+'); @@ -186,6 +188,7 @@ public async Task CanUseEncoderWithServiceBusMessage() var employee = new Employee { Age = 42, Name = "Caketown" }; ServiceBusMessage message = await encoder.EncodeMessageDataAsync(employee); + Assert.IsFalse(((MessageWithMetadata) message).IsReadOnly); string[] contentType = message.ContentType.Split('+'); Assert.AreEqual(2, contentType.Length); @@ -194,6 +197,7 @@ public async Task CanUseEncoderWithServiceBusMessage() ServiceBusReceivedMessage receivedMessage = ServiceBusModelFactory.ServiceBusReceivedMessage(body: message.Body, contentType: message.ContentType); + Assert.IsTrue(((MessageWithMetadata) receivedMessage).IsReadOnly); Employee deserialized = (Employee)await encoder.DecodeMessageDataAsync(receivedMessage, typeof(Employee)); diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/Azure.Messaging.ServiceBus.sln b/sdk/servicebus/Azure.Messaging.ServiceBus/Azure.Messaging.ServiceBus.sln index 915aafa294258..0fed414342f8f 100644 --- a/sdk/servicebus/Azure.Messaging.ServiceBus/Azure.Messaging.ServiceBus.sln +++ b/sdk/servicebus/Azure.Messaging.ServiceBus/Azure.Messaging.ServiceBus.sln @@ -42,6 +42,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Azure.Messaging.ServiceBus. EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Azure.Test.Perf", "..\..\..\common\Perf\Azure.Test.Perf\Azure.Test.Perf.csproj", "{A6A7CEAE-C1F9-4B9B-A19F-4AA38ADF5209}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Azure.Core.Experimental", "..\..\core\Azure.Core.Experimental\src\Azure.Core.Experimental.csproj", "{B42578D2-76CD-4F4E-9999-CDB466477488}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -80,6 +82,10 @@ Global {A6A7CEAE-C1F9-4B9B-A19F-4AA38ADF5209}.Debug|Any CPU.Build.0 = Debug|Any CPU {A6A7CEAE-C1F9-4B9B-A19F-4AA38ADF5209}.Release|Any CPU.ActiveCfg = Release|Any CPU {A6A7CEAE-C1F9-4B9B-A19F-4AA38ADF5209}.Release|Any CPU.Build.0 = Release|Any CPU + {B42578D2-76CD-4F4E-9999-CDB466477488}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {B42578D2-76CD-4F4E-9999-CDB466477488}.Debug|Any CPU.Build.0 = Debug|Any CPU + {B42578D2-76CD-4F4E-9999-CDB466477488}.Release|Any CPU.ActiveCfg = Release|Any CPU + {B42578D2-76CD-4F4E-9999-CDB466477488}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/api/Azure.Messaging.ServiceBus.netstandard2.0.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/api/Azure.Messaging.ServiceBus.netstandard2.0.cs index 8a2251c75ca02..02be6add985fd 100644 --- a/sdk/servicebus/Azure.Messaging.ServiceBus/api/Azure.Messaging.ServiceBus.netstandard2.0.cs +++ b/sdk/servicebus/Azure.Messaging.ServiceBus/api/Azure.Messaging.ServiceBus.netstandard2.0.cs @@ -165,6 +165,8 @@ public ServiceBusMessage(string body) { } public string CorrelationId { get { throw null; } set { } } [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] public override System.BinaryData Data { get { throw null; } set { } } + [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] + public override bool IsReadOnly { get { throw null; } } public string MessageId { get { throw null; } set { } } public string PartitionKey { get { throw null; } set { } } public string ReplyTo { get { throw null; } set { } } @@ -273,6 +275,8 @@ internal ServiceBusReceivedMessage() { } public long EnqueuedSequenceNumber { get { throw null; } } public System.DateTimeOffset EnqueuedTime { get { throw null; } } public System.DateTimeOffset ExpiresAt { get { throw null; } } + [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] + public override bool IsReadOnly { get { throw null; } } public System.DateTimeOffset LockedUntil { get { throw null; } } public string LockToken { get { throw null; } } public string MessageId { get { throw null; } } diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusMessage.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusMessage.cs index 6413ee5920cb9..ea0323c37bce7 100755 --- a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusMessage.cs +++ b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusMessage.cs @@ -373,6 +373,13 @@ public override string ContentType } } + /// + /// Hidden property that indicates that the is not read-only. This is part of + /// the abstraction. + /// + [EditorBrowsable(EditorBrowsableState.Never)] + public override bool IsReadOnly => false; + /// Gets or sets the address of an entity to send replies to. /// The reply entity address. /// diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusReceivedMessage.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusReceivedMessage.cs index 2d4ff977e8f57..00f0809a64975 100644 --- a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusReceivedMessage.cs +++ b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusReceivedMessage.cs @@ -185,6 +185,13 @@ public override string ContentType set => throw new NotImplementedException("Content type cannot be set on a ServiceBusReceivedMessage"); } + /// + /// Hidden property that indicates that the is read-only. This is part of + /// the abstraction. + /// + [EditorBrowsable(EditorBrowsableState.Never)] + public override bool IsReadOnly => true; + /// Gets the address of an entity to send replies to. /// The reply entity address. /// From 6d10e3aaaa890d5ba00d2741158223d3b1c8e461 Mon Sep 17 00:00:00 2001 From: JoshLove-msft <54595583+JoshLove-msft@users.noreply.github.com> Date: Tue, 30 Nov 2021 17:04:24 -0800 Subject: [PATCH 4/6] remove IMessageWithContentType --- .../Azure.Core.Experimental.netstandard2.0.cs | 5 -- .../src/IMessageWithContentType.cs | 23 -------- ...g.EventHubs.Experimental.netstandard2.0.cs | 58 +++++++++++++++++++ 3 files changed, 58 insertions(+), 28 deletions(-) delete mode 100644 sdk/core/Azure.Core.Experimental/src/IMessageWithContentType.cs create mode 100644 sdk/eventhub/Azure.Messaging.EventHubs.Experimental/api/Azure.Messaging.EventHubs.Experimental.netstandard2.0.cs diff --git a/sdk/core/Azure.Core.Experimental/api/Azure.Core.Experimental.netstandard2.0.cs b/sdk/core/Azure.Core.Experimental/api/Azure.Core.Experimental.netstandard2.0.cs index a224f37a6f6cb..b5dc6a22b536a 100644 --- a/sdk/core/Azure.Core.Experimental/api/Azure.Core.Experimental.netstandard2.0.cs +++ b/sdk/core/Azure.Core.Experimental/api/Azure.Core.Experimental.netstandard2.0.cs @@ -88,11 +88,6 @@ public void Set(string propertyName, string? value) { } } namespace Azure.Messaging { - public partial interface IMessageWithContentType - { - string ContentType { get; set; } - System.BinaryData Data { get; set; } - } public abstract partial class MessageWithMetadata { protected MessageWithMetadata() { } diff --git a/sdk/core/Azure.Core.Experimental/src/IMessageWithContentType.cs b/sdk/core/Azure.Core.Experimental/src/IMessageWithContentType.cs deleted file mode 100644 index 1c895c3fe0a0e..0000000000000 --- a/sdk/core/Azure.Core.Experimental/src/IMessageWithContentType.cs +++ /dev/null @@ -1,23 +0,0 @@ -// Copyright (c) Microsoft Corporation. All rights reserved. -// Licensed under the MIT License. - -using System; - -namespace Azure.Messaging -{ - /// - /// An interface for a message containing a content type along with its data. - /// - public interface IMessageWithContentType - { - /// - /// Gets or sets the message data. - /// - BinaryData Data { get; set; } - - /// - /// Gets or sets the message content type. - /// - string ContentType { get; set; } - } -} \ No newline at end of file diff --git a/sdk/eventhub/Azure.Messaging.EventHubs.Experimental/api/Azure.Messaging.EventHubs.Experimental.netstandard2.0.cs b/sdk/eventhub/Azure.Messaging.EventHubs.Experimental/api/Azure.Messaging.EventHubs.Experimental.netstandard2.0.cs new file mode 100644 index 0000000000000..d09a7317f5c13 --- /dev/null +++ b/sdk/eventhub/Azure.Messaging.EventHubs.Experimental/api/Azure.Messaging.EventHubs.Experimental.netstandard2.0.cs @@ -0,0 +1,58 @@ +namespace Azure.Messaging.EventHubs.Producer +{ + public partial class IdempotentProducer : Azure.Messaging.EventHubs.Producer.EventHubProducerClient + { + protected IdempotentProducer() { } + public IdempotentProducer(Azure.Messaging.EventHubs.EventHubConnection connection, Azure.Messaging.EventHubs.Producer.IdempotentProducerOptions clientOptions = null) { } + public IdempotentProducer(string connectionString) { } + public IdempotentProducer(string connectionString, Azure.Messaging.EventHubs.Producer.IdempotentProducerOptions clientOptions) { } + public IdempotentProducer(string connectionString, string eventHubName) { } + public IdempotentProducer(string fullyQualifiedNamespace, string eventHubName, Azure.AzureNamedKeyCredential credential, Azure.Messaging.EventHubs.Producer.IdempotentProducerOptions clientOptions = null) { } + public IdempotentProducer(string fullyQualifiedNamespace, string eventHubName, Azure.AzureSasCredential credential, Azure.Messaging.EventHubs.Producer.IdempotentProducerOptions clientOptions = null) { } + public IdempotentProducer(string fullyQualifiedNamespace, string eventHubName, Azure.Core.TokenCredential credential, Azure.Messaging.EventHubs.Producer.IdempotentProducerOptions clientOptions = null) { } + public IdempotentProducer(string connectionString, string eventHubName, Azure.Messaging.EventHubs.Producer.IdempotentProducerOptions clientOptions) { } + public virtual System.Threading.Tasks.Task GetPartitionPublishingPropertiesAsync(string partitionId, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; } + } + public partial class IdempotentProducerOptions + { + public IdempotentProducerOptions() { } + public Azure.Messaging.EventHubs.EventHubConnectionOptions ConnectionOptions { get { throw null; } set { } } + public bool EnableIdempotentPartitions { get { throw null; } set { } } + public string Identifier { get { throw null; } set { } } + public System.Collections.Generic.Dictionary PartitionOptions { get { throw null; } } + public Azure.Messaging.EventHubs.EventHubsRetryOptions RetryOptions { get { throw null; } set { } } + [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] + public override bool Equals(object obj) { throw null; } + [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] + public override int GetHashCode() { throw null; } + [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] + public override string ToString() { throw null; } + } + public partial class PartitionPublishingOptions + { + public PartitionPublishingOptions() { } + public short? OwnerLevel { get { throw null; } set { } } + public long? ProducerGroupId { get { throw null; } set { } } + public int? StartingSequenceNumber { get { throw null; } set { } } + [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] + public override bool Equals(object obj) { throw null; } + [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] + public override int GetHashCode() { throw null; } + [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] + public override string ToString() { throw null; } + } + public partial class PartitionPublishingProperties + { + public PartitionPublishingProperties(bool isIdempotentPublishingEnabled, long? producerGroupId, short? ownerLevel, int? lastPublishedSequenceNumber) { } + public bool IsIdempotentPublishingEnabled { get { throw null; } } + public int? LastPublishedSequenceNumber { get { throw null; } } + public short? OwnerLevel { get { throw null; } } + public long? ProducerGroupId { get { throw null; } } + [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] + public override bool Equals(object obj) { throw null; } + [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] + public override int GetHashCode() { throw null; } + [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] + public override string ToString() { throw null; } + } +} From 0c6cc352e43d7982d4527fef8373a6edbfaf367b Mon Sep 17 00:00:00 2001 From: JoshLove-msft <54595583+JoshLove-msft@users.noreply.github.com> Date: Tue, 30 Nov 2021 17:05:08 -0800 Subject: [PATCH 5/6] delete commented test --- .../tests/Samples/Sample01_ReadmeSnippets.cs | 23 ------------------- 1 file changed, 23 deletions(-) diff --git a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/Samples/Sample01_ReadmeSnippets.cs b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/Samples/Sample01_ReadmeSnippets.cs index 36c1b562968d8..982a7b9fa90dc 100644 --- a/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/Samples/Sample01_ReadmeSnippets.cs +++ b/sdk/schemaregistry/Microsoft.Azure.Data.SchemaRegistry.ApacheAvro/tests/Samples/Sample01_ReadmeSnippets.cs @@ -33,28 +33,5 @@ public void CreateSchemaRegistryClient() this.schemaRegistryClient = schemaRegistryClient; } - - // [Test] - // public void EncodeDecodeEventData() - // { - // string groupName = TestEnvironment.SchemaRegistryGroup; - // - // #region Snippet:SchemaRegistryAvroEncodeEventData - // var employee = new Employee { Age = 42, Name = "John Doe" }; - // var encoder = new SchemaRegistryAvroEncoder(schemaRegistryClient, groupName, new SchemaRegistryAvroObjectEncoderOptions { AutoRegisterSchemas = true }); - // var eventData = new EventData(); - // - // encoder.EncodeMessageData(eventData, employee, typeof(Employee)); - // - // // the schema Id will be included as a parameter of the content type - // Console.WriteLine(eventData.ContentType); - // - // // the serialized Avro data will be stored in the EventBody - // Console.WriteLine(eventData.EventBody); - // - // // We can also get the Employee model back out from the serialized data - // employee = (Employee) encoder.DecodeMessageData(eventData, typeof(Employee)); - // #endregion - // } } } From fabd400ea81cb724f7f5036e9560e9722ebf5dbb Mon Sep 17 00:00:00 2001 From: JoshLove-msft <54595583+JoshLove-msft@users.noreply.github.com> Date: Tue, 30 Nov 2021 17:08:27 -0800 Subject: [PATCH 6/6] Remove EventHubs Experimental API file --- ...g.EventHubs.Experimental.netstandard2.0.cs | 58 ------------------- 1 file changed, 58 deletions(-) delete mode 100644 sdk/eventhub/Azure.Messaging.EventHubs.Experimental/api/Azure.Messaging.EventHubs.Experimental.netstandard2.0.cs diff --git a/sdk/eventhub/Azure.Messaging.EventHubs.Experimental/api/Azure.Messaging.EventHubs.Experimental.netstandard2.0.cs b/sdk/eventhub/Azure.Messaging.EventHubs.Experimental/api/Azure.Messaging.EventHubs.Experimental.netstandard2.0.cs deleted file mode 100644 index d09a7317f5c13..0000000000000 --- a/sdk/eventhub/Azure.Messaging.EventHubs.Experimental/api/Azure.Messaging.EventHubs.Experimental.netstandard2.0.cs +++ /dev/null @@ -1,58 +0,0 @@ -namespace Azure.Messaging.EventHubs.Producer -{ - public partial class IdempotentProducer : Azure.Messaging.EventHubs.Producer.EventHubProducerClient - { - protected IdempotentProducer() { } - public IdempotentProducer(Azure.Messaging.EventHubs.EventHubConnection connection, Azure.Messaging.EventHubs.Producer.IdempotentProducerOptions clientOptions = null) { } - public IdempotentProducer(string connectionString) { } - public IdempotentProducer(string connectionString, Azure.Messaging.EventHubs.Producer.IdempotentProducerOptions clientOptions) { } - public IdempotentProducer(string connectionString, string eventHubName) { } - public IdempotentProducer(string fullyQualifiedNamespace, string eventHubName, Azure.AzureNamedKeyCredential credential, Azure.Messaging.EventHubs.Producer.IdempotentProducerOptions clientOptions = null) { } - public IdempotentProducer(string fullyQualifiedNamespace, string eventHubName, Azure.AzureSasCredential credential, Azure.Messaging.EventHubs.Producer.IdempotentProducerOptions clientOptions = null) { } - public IdempotentProducer(string fullyQualifiedNamespace, string eventHubName, Azure.Core.TokenCredential credential, Azure.Messaging.EventHubs.Producer.IdempotentProducerOptions clientOptions = null) { } - public IdempotentProducer(string connectionString, string eventHubName, Azure.Messaging.EventHubs.Producer.IdempotentProducerOptions clientOptions) { } - public virtual System.Threading.Tasks.Task GetPartitionPublishingPropertiesAsync(string partitionId, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; } - } - public partial class IdempotentProducerOptions - { - public IdempotentProducerOptions() { } - public Azure.Messaging.EventHubs.EventHubConnectionOptions ConnectionOptions { get { throw null; } set { } } - public bool EnableIdempotentPartitions { get { throw null; } set { } } - public string Identifier { get { throw null; } set { } } - public System.Collections.Generic.Dictionary PartitionOptions { get { throw null; } } - public Azure.Messaging.EventHubs.EventHubsRetryOptions RetryOptions { get { throw null; } set { } } - [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] - public override bool Equals(object obj) { throw null; } - [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] - public override int GetHashCode() { throw null; } - [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] - public override string ToString() { throw null; } - } - public partial class PartitionPublishingOptions - { - public PartitionPublishingOptions() { } - public short? OwnerLevel { get { throw null; } set { } } - public long? ProducerGroupId { get { throw null; } set { } } - public int? StartingSequenceNumber { get { throw null; } set { } } - [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] - public override bool Equals(object obj) { throw null; } - [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] - public override int GetHashCode() { throw null; } - [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] - public override string ToString() { throw null; } - } - public partial class PartitionPublishingProperties - { - public PartitionPublishingProperties(bool isIdempotentPublishingEnabled, long? producerGroupId, short? ownerLevel, int? lastPublishedSequenceNumber) { } - public bool IsIdempotentPublishingEnabled { get { throw null; } } - public int? LastPublishedSequenceNumber { get { throw null; } } - public short? OwnerLevel { get { throw null; } } - public long? ProducerGroupId { get { throw null; } } - [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] - public override bool Equals(object obj) { throw null; } - [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] - public override int GetHashCode() { throw null; } - [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)] - public override string ToString() { throw null; } - } -}