From 8d25615dd9e7d852d6d92cb424378b04ad3559d2 Mon Sep 17 00:00:00 2001 From: Erik OLeary Date: Thu, 6 Apr 2023 15:51:03 -0500 Subject: [PATCH] Add system text json support for actor serialization Signed-off-by: Erik O'Leary --- src/Dapr.Actors/Client/ActorProxyFactory.cs | 11 +- src/Dapr.Actors/Client/ActorProxyOptions.cs | 5 + ...geBodyDataContractSerializationProvider.cs | 17 +- .../ActorMessageBodyJsonConverter.cs | 100 ++++++++++ ...torMessageBodyJsonSerializationProvider.cs | 185 ++++++++++++++++++ .../IActorRequestMessageBodySerializer.cs | 3 +- .../IActorResponseMessageBodySerializer.cs | 3 +- src/Dapr.Actors/DaprHttpInteractor.cs | 18 +- src/Dapr.Actors/Runtime/ActorManager.cs | 15 +- src/Dapr.Actors/Runtime/ActorRuntime.cs | 1 + .../Runtime/ActorRuntimeOptions.cs | 20 +- .../Runtime/DataContractStateSerializer.cs | 82 -------- .../Runtime/ActorManagerTests.cs | 4 +- .../ISerializationActor.cs | 22 +++ .../Actors/SerializationActor.cs | 26 +++ test/Dapr.E2E.Test.App/Startup.cs | 6 + .../Actors/E2ETests.CustomSerializerTests.cs | 88 +++++++++ test/Dapr.E2E.Test/DaprRunConfiguration.cs | 6 +- test/Dapr.E2E.Test/DaprTestApp.cs | 5 + 19 files changed, 506 insertions(+), 111 deletions(-) create mode 100644 src/Dapr.Actors/Communication/ActorMessageBodyJsonConverter.cs create mode 100644 src/Dapr.Actors/Communication/ActorMessageBodyJsonSerializationProvider.cs delete mode 100644 src/Dapr.Actors/Runtime/DataContractStateSerializer.cs create mode 100644 test/Dapr.E2E.Test.Actors/ISerializationActor.cs create mode 100644 test/Dapr.E2E.Test.App/Actors/SerializationActor.cs create mode 100644 test/Dapr.E2E.Test/Actors/E2ETests.CustomSerializerTests.cs diff --git a/src/Dapr.Actors/Client/ActorProxyFactory.cs b/src/Dapr.Actors/Client/ActorProxyFactory.cs index 9fd5edddb..4a8fe3a08 100644 --- a/src/Dapr.Actors/Client/ActorProxyFactory.cs +++ b/src/Dapr.Actors/Client/ActorProxyFactory.cs @@ -16,6 +16,7 @@ namespace Dapr.Actors.Client using System; using System.Net.Http; using Dapr.Actors.Builder; + using Dapr.Actors.Communication; using Dapr.Actors.Communication.Client; /// @@ -79,7 +80,15 @@ public object CreateActorProxy(ActorId actorId, Type actorInterfaceType, string options ??= this.DefaultOptions; var daprInteractor = new DaprHttpInteractor(this.handler, options.HttpEndpoint, options.DaprApiToken, options.RequestTimeout); - var remotingClient = new ActorRemotingClient(daprInteractor); + + // provide a serializer if 'useJsonSerialization' is true and no serialization provider is provided. + IActorMessageBodySerializationProvider serializationProvider = null; + if (options.UseJsonSerialization) + { + serializationProvider = new ActorMessageBodyJsonSerializationProvider(options.JsonSerializerOptions); + } + + var remotingClient = new ActorRemotingClient(daprInteractor, serializationProvider); var proxyGenerator = ActorCodeBuilder.GetOrCreateProxyGenerator(actorInterfaceType); var actorProxy = proxyGenerator.CreateActorProxy(); actorProxy.Initialize(remotingClient, actorId, actorType, options); diff --git a/src/Dapr.Actors/Client/ActorProxyOptions.cs b/src/Dapr.Actors/Client/ActorProxyOptions.cs index 808605c70..665a1dced 100644 --- a/src/Dapr.Actors/Client/ActorProxyOptions.cs +++ b/src/Dapr.Actors/Client/ActorProxyOptions.cs @@ -62,5 +62,10 @@ public JsonSerializerOptions JsonSerializerOptions /// The timeout allowed for an actor request. Can be set to System.Threading.Timeout.InfiniteTimeSpan to disable any timeouts. /// public TimeSpan? RequestTimeout { get; set; } = null; + + /// + /// Enable JSON serialization for actor proxy message serialization in both remoting and non-remoting invocations. + /// + public bool UseJsonSerialization { get; set; } } } diff --git a/src/Dapr.Actors/Communication/ActorMessageBodyDataContractSerializationProvider.cs b/src/Dapr.Actors/Communication/ActorMessageBodyDataContractSerializationProvider.cs index e1991df26..cf16ee2d8 100644 --- a/src/Dapr.Actors/Communication/ActorMessageBodyDataContractSerializationProvider.cs +++ b/src/Dapr.Actors/Communication/ActorMessageBodyDataContractSerializationProvider.cs @@ -17,6 +17,7 @@ namespace Dapr.Actors.Communication using System.Collections.Generic; using System.IO; using System.Runtime.Serialization; + using System.Threading.Tasks; using System.Xml; /// @@ -185,21 +186,21 @@ byte[] IActorRequestMessageBodySerializer.Serialize(IActorRequestMessageBody act return stream.ToArray(); } - IActorRequestMessageBody IActorRequestMessageBodySerializer.Deserialize(Stream stream) + ValueTask IActorRequestMessageBodySerializer.DeserializeAsync(Stream stream) { if (stream == null) { - return null; + return default; } if (stream.Length == 0) { - return null; + return default; } stream.Position = 0; using var reader = this.CreateXmlDictionaryReader(stream); - return (TRequest)this.serializer.ReadObject(reader); + return new ValueTask((TRequest)this.serializer.ReadObject(reader)); } byte[] IActorResponseMessageBodySerializer.Serialize(IActorResponseMessageBody actorResponseMessageBody) @@ -217,11 +218,11 @@ byte[] IActorResponseMessageBodySerializer.Serialize(IActorResponseMessageBody a return stream.ToArray(); } - IActorResponseMessageBody IActorResponseMessageBodySerializer.Deserialize(Stream messageBody) + ValueTask IActorResponseMessageBodySerializer.DeserializeAsync(Stream messageBody) { if (messageBody == null) { - return null; + return default; } // TODO check performance @@ -231,11 +232,11 @@ IActorResponseMessageBody IActorResponseMessageBodySerializer.Deserialize(Stream if (stream.Capacity == 0) { - return null; + return default; } using var reader = this.CreateXmlDictionaryReader(stream); - return (TResponse)this.serializer.ReadObject(reader); + return new ValueTask((TResponse)this.serializer.ReadObject(reader)); } /// diff --git a/src/Dapr.Actors/Communication/ActorMessageBodyJsonConverter.cs b/src/Dapr.Actors/Communication/ActorMessageBodyJsonConverter.cs new file mode 100644 index 000000000..70bdfe05f --- /dev/null +++ b/src/Dapr.Actors/Communication/ActorMessageBodyJsonConverter.cs @@ -0,0 +1,100 @@ +// ------------------------------------------------------------------------ +// Copyright 2021 The Dapr Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ------------------------------------------------------------------------ + +using System; +using System.Collections.Generic; +using System.Text.Json; +using System.Text.Json.Serialization; + +namespace Dapr.Actors.Communication +{ + internal class ActorMessageBodyJsonConverter : JsonConverter + { + private readonly List methodRequestParameterTypes; + private readonly List wrappedRequestMessageTypes; + private readonly Type wrapperMessageType; + + public ActorMessageBodyJsonConverter( + List methodRequestParameterTypes, + List wrappedRequestMessageTypes = null + ) + { + this.methodRequestParameterTypes = methodRequestParameterTypes; + this.wrappedRequestMessageTypes = wrappedRequestMessageTypes; + + if (this.wrappedRequestMessageTypes != null && this.wrappedRequestMessageTypes.Count == 1) + { + this.wrapperMessageType = this.wrappedRequestMessageTypes[0]; + } + + // If T is of WrappedMessageBody, then get the 'Value' property. + if (typeof(T).IsAssignableFrom(typeof(WrappedMessageBody))) + { + } + } + + public override T Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options) + { + // Ensure start-of-object, then advance + if (reader.TokenType != JsonTokenType.StartObject) throw new JsonException(); + reader.Read(); + + // Ensure property name, then advance + if (reader.TokenType != JsonTokenType.PropertyName || reader.GetString() != "value") throw new JsonException(); + reader.Read(); + + // If the value is null, return null. + if (reader.TokenType == JsonTokenType.Null) + { + // Read the end object token. + reader.Read(); + return default; + } + + // If the value is an object, deserialize it to wrapper message type + if (this.wrapperMessageType != null) + { + var value = JsonSerializer.Deserialize(ref reader, this.wrapperMessageType, options); + + // Construct a new WrappedMessageBody with the deserialized value. + var wrapper = new WrappedMessageBody() + { + Value = value, + }; + + // Read the end object token. + reader.Read(); + + // Coerce the type to T; required because WrappedMessageBody inherits from two separate interfaces, which + // cannot both be used as generic constraints + return (T)((object)wrapper); + } + + return JsonSerializer.Deserialize(ref reader, options); + } + + public override void Write(Utf8JsonWriter writer, T value, JsonSerializerOptions options) + { + writer.WriteStartObject(); + writer.WritePropertyName("value"); + + if (value is WrappedMessageBody body) + { + JsonSerializer.Serialize(writer, body.Value, body.Value.GetType(), options); + } + else + writer.WriteNullValue(); + writer.WriteEndObject(); + } + } +} diff --git a/src/Dapr.Actors/Communication/ActorMessageBodyJsonSerializationProvider.cs b/src/Dapr.Actors/Communication/ActorMessageBodyJsonSerializationProvider.cs new file mode 100644 index 000000000..ff39848d1 --- /dev/null +++ b/src/Dapr.Actors/Communication/ActorMessageBodyJsonSerializationProvider.cs @@ -0,0 +1,185 @@ +// ------------------------------------------------------------------------ +// Copyright 2021 The Dapr Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ------------------------------------------------------------------------ + +namespace Dapr.Actors.Communication +{ + using System; + using System.Collections.Generic; + using System.IO; + using System.Text.Json; + using System.Threading.Tasks; + using System.Xml; + + /// + /// This is the implmentation for used by remoting service and client during + /// request/response serialization . It uses request Wrapping and data contract for serialization. + /// + internal class ActorMessageBodyJsonSerializationProvider : IActorMessageBodySerializationProvider + { + public JsonSerializerOptions Options { get; } + + /// + /// Initializes a new instance of the class. + /// + public ActorMessageBodyJsonSerializationProvider(JsonSerializerOptions options) + { + Options = options; + } + + /// + /// Creates a MessageFactory for Wrapped Message Json Remoting Types. This is used to create Remoting Request/Response objects. + /// + /// + /// that provides an instance of the factory for creating + /// remoting request and response message bodies. + /// + public IActorMessageBodyFactory CreateMessageBodyFactory() + { + return new WrappedRequestMessageFactory(); + } + + /// + /// Creates IActorRequestMessageBodySerializer for a serviceInterface using Wrapped Message Json implementation. + /// + /// The remoted service interface. + /// The union of parameter types of all of the methods of the specified interface. + /// Wrapped Request Types for all Methods. + /// + /// An instance of the that can serialize the service + /// actor request message body to a messaging body for transferring over the transport. + /// + public IActorRequestMessageBodySerializer CreateRequestMessageBodySerializer( + Type serviceInterfaceType, + IEnumerable methodRequestParameterTypes, + IEnumerable wrappedRequestMessageTypes = null) + { + return new MemoryStreamMessageBodySerializer(Options, serviceInterfaceType, methodRequestParameterTypes, wrappedRequestMessageTypes); + } + + /// + /// Creates IActorResponseMessageBodySerializer for a serviceInterface using Wrapped Message Json implementation. + /// + /// The remoted service interface. + /// The return types of all of the methods of the specified interface. + /// Wrapped Response Types for all remoting methods. + /// + /// An instance of the that can serialize the service + /// actor response message body to a messaging body for transferring over the transport. + /// + public IActorResponseMessageBodySerializer CreateResponseMessageBodySerializer( + Type serviceInterfaceType, + IEnumerable methodReturnTypes, + IEnumerable wrappedResponseMessageTypes = null) + { + return new MemoryStreamMessageBodySerializer(Options, serviceInterfaceType, methodReturnTypes, wrappedResponseMessageTypes); + } + + /// + /// Default serializer for service remoting request and response message body that uses the + /// memory stream to create outgoing message buffers. + /// + private class MemoryStreamMessageBodySerializer : + IActorRequestMessageBodySerializer, + IActorResponseMessageBodySerializer + where TRequest : IActorRequestMessageBody + where TResponse : IActorResponseMessageBody + { + private readonly JsonSerializerOptions serializerOptions; + + public MemoryStreamMessageBodySerializer( + JsonSerializerOptions serializerOptions, + Type serviceInterfaceType, + IEnumerable methodRequestParameterTypes, + IEnumerable wrappedRequestMessageTypes = null) + { + var _methodRequestParameterTypes = new List(methodRequestParameterTypes); + var _wrappedRequestMessageTypes = new List(wrappedRequestMessageTypes); + + this.serializerOptions = new(serializerOptions) + { + // Workaround since WrappedMessageBody creates an object + // with parameters as fields + IncludeFields = true, + }; + + this.serializerOptions.Converters.Add(new ActorMessageBodyJsonConverter(_methodRequestParameterTypes, _wrappedRequestMessageTypes)); + this.serializerOptions.Converters.Add(new ActorMessageBodyJsonConverter(_methodRequestParameterTypes, _wrappedRequestMessageTypes)); + } + + byte[] IActorRequestMessageBodySerializer.Serialize(IActorRequestMessageBody actorRequestMessageBody) + { + if (actorRequestMessageBody == null) + { + return null; + } + + using var stream = new MemoryStream(); + using var writer = new Utf8JsonWriter(stream); + JsonSerializer.Serialize(writer, actorRequestMessageBody, this.serializerOptions); + writer.Flush(); + + return stream.ToArray(); + } + + async ValueTask IActorRequestMessageBodySerializer.DeserializeAsync(Stream stream) + { + if (stream == null) + { + return default; + } + + if (stream.Length == 0) + { + return default; + } + + stream.Position = 0; + return await JsonSerializer.DeserializeAsync(stream, this.serializerOptions); + } + + byte[] IActorResponseMessageBodySerializer.Serialize(IActorResponseMessageBody actorResponseMessageBody) + { + if (actorResponseMessageBody == null) + { + return null; + } + + using var stream = new MemoryStream(); + using var writer = new Utf8JsonWriter(stream); + JsonSerializer.Serialize(writer, actorResponseMessageBody, this.serializerOptions); + writer.Flush(); + + return stream.ToArray(); + } + + async ValueTask IActorResponseMessageBodySerializer.DeserializeAsync(Stream messageBody) + { + if (messageBody == null) + { + return null; + } + + using var stream = new MemoryStream(); + messageBody.CopyTo(stream); + stream.Position = 0; + + if (stream.Capacity == 0) + { + return null; + } + + return await JsonSerializer.DeserializeAsync(stream, this.serializerOptions); + } + } + } +} diff --git a/src/Dapr.Actors/Communication/IActorRequestMessageBodySerializer.cs b/src/Dapr.Actors/Communication/IActorRequestMessageBodySerializer.cs index 66c6abb82..b58257381 100644 --- a/src/Dapr.Actors/Communication/IActorRequestMessageBodySerializer.cs +++ b/src/Dapr.Actors/Communication/IActorRequestMessageBodySerializer.cs @@ -14,6 +14,7 @@ namespace Dapr.Actors.Communication { using System.IO; + using System.Threading.Tasks; /// /// Defines the interface that must be implemented to provide a serializer/deserializer for remoting request message body. @@ -32,6 +33,6 @@ internal interface IActorRequestMessageBodySerializer /// /// Serialized message body. /// Deserialized remoting request message body object. - IActorRequestMessageBody Deserialize(Stream messageBody); + ValueTask DeserializeAsync(Stream messageBody); } } diff --git a/src/Dapr.Actors/Communication/IActorResponseMessageBodySerializer.cs b/src/Dapr.Actors/Communication/IActorResponseMessageBodySerializer.cs index 2c31019ba..b54191f91 100644 --- a/src/Dapr.Actors/Communication/IActorResponseMessageBodySerializer.cs +++ b/src/Dapr.Actors/Communication/IActorResponseMessageBodySerializer.cs @@ -14,6 +14,7 @@ namespace Dapr.Actors.Communication { using System.IO; + using System.Threading.Tasks; /// /// Defines the interface that must be implemented to provide a serializer/deserializer for actor response message body. @@ -32,6 +33,6 @@ internal interface IActorResponseMessageBodySerializer /// /// Serialized message body. /// Deserialized actor response message body object. - IActorResponseMessageBody Deserialize(Stream messageBody); + ValueTask DeserializeAsync(Stream messageBody); } } diff --git a/src/Dapr.Actors/DaprHttpInteractor.cs b/src/Dapr.Actors/DaprHttpInteractor.cs index 410925dae..4695375fb 100644 --- a/src/Dapr.Actors/DaprHttpInteractor.cs +++ b/src/Dapr.Actors/DaprHttpInteractor.cs @@ -1,4 +1,4 @@ -// ------------------------------------------------------------------------ +// ------------------------------------------------------------------------ // Copyright 2021 The Dapr Authors // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -70,7 +70,7 @@ HttpRequestMessage RequestFunc() return request; } - var response = await this.SendAsync(RequestFunc, relativeUrl, cancellationToken); + using var response = await this.SendAsync(RequestFunc, relativeUrl, cancellationToken); var stringResponse = await response.Content.ReadAsStringAsync(); return stringResponse; } @@ -164,11 +164,11 @@ HttpRequestMessage RequestFunc() // actorResponseMessageHeader is not null, it means there is remote exception if (actorResponseMessageHeader != null) { - var isDeserialzied = + var isDeserialized = ActorInvokeException.ToException( responseMessageBody, out var remoteMethodException); - if (isDeserialzied) + if (isDeserialized) { var exceptionDetails = GetExceptionDetails(header.ToString()); throw new ActorMethodInvocationException( @@ -185,7 +185,7 @@ HttpRequestMessage RequestFunc() } } - actorResponseMessageBody = responseBodySerializer.Deserialize(responseMessageBody); + actorResponseMessageBody = await responseBodySerializer.DeserializeAsync(responseMessageBody); } return new ActorResponseMessage(actorResponseMessageHeader, actorResponseMessageBody); @@ -231,8 +231,8 @@ HttpRequestMessage RequestFunc() } var response = await this.SendAsync(RequestFunc, relativeUrl, cancellationToken); - var byteArray = await response.Content.ReadAsStreamAsync(); - return byteArray; + var stream = await response.Content.ReadAsStreamAsync(); + return stream; } public Task RegisterReminderAsync(string actorType, string actorId, string reminderName, string data, CancellationToken cancellationToken = default) @@ -351,7 +351,7 @@ internal async Task SendAsyncGetResponseAsRawJson( string relativeUri, CancellationToken cancellationToken) { - var response = await this.SendAsyncHandleUnsuccessfulResponse(requestFunc, relativeUri, cancellationToken); + using var response = await this.SendAsyncHandleUnsuccessfulResponse(requestFunc, relativeUri, cancellationToken); var retValue = default(string); if (response != null && response.Content != null) @@ -458,7 +458,7 @@ private async Task SendAsyncHandleSecurityExceptions( HttpResponseMessage response; // Get the request using the Func as same request cannot be resent when retries are implemented. - var request = requestFunc.Invoke(); + using var request = requestFunc.Invoke(); // add token for dapr api token based authentication this.AddDaprApiTokenHeader(request); diff --git a/src/Dapr.Actors/Runtime/ActorManager.cs b/src/Dapr.Actors/Runtime/ActorManager.cs index e7f29f5cc..adbb127c9 100644 --- a/src/Dapr.Actors/Runtime/ActorManager.cs +++ b/src/Dapr.Actors/Runtime/ActorManager.cs @@ -56,7 +56,8 @@ internal sealed class ActorManager internal ActorManager( ActorRegistration registration, ActorActivator activator, - JsonSerializerOptions jsonSerializerOptions, + JsonSerializerOptions jsonSerializerOptions, + bool useJsonSerialization, ILoggerFactory loggerFactory, IActorProxyFactory proxyFactory, IDaprInteractor daprInteractor) @@ -78,7 +79,15 @@ internal ActorManager( this.activeActors = new ConcurrentDictionary(); this.reminderMethodContext = ActorMethodContext.CreateForReminder(ReceiveReminderMethodName); this.timerMethodContext = ActorMethodContext.CreateForTimer(TimerMethodName); - this.serializersManager = IntializeSerializationManager(null); + + // provide a serializer if 'useJsonSerialization' is true and no serialization provider is provided. + IActorMessageBodySerializationProvider serializationProvider = null; + if (useJsonSerialization) + { + serializationProvider = new ActorMessageBodyJsonSerializationProvider(jsonSerializerOptions); + } + + this.serializersManager = IntializeSerializationManager(serializationProvider); this.messageBodyFactory = new WrappedRequestMessageFactory(); this.logger = loggerFactory.CreateLogger(this.GetType()); @@ -103,7 +112,7 @@ internal async Task> DispatchWithRemotingAsync(ActorId act using (var stream = new MemoryStream()) { await data.CopyToAsync(stream); - actorMessageBody = msgBodySerializer.Deserialize(stream); + actorMessageBody = await msgBodySerializer.DeserializeAsync(stream); } // Call the method on the method dispatcher using the Func below. diff --git a/src/Dapr.Actors/Runtime/ActorRuntime.cs b/src/Dapr.Actors/Runtime/ActorRuntime.cs index 8d2ae0cab..7f01fefb7 100644 --- a/src/Dapr.Actors/Runtime/ActorRuntime.cs +++ b/src/Dapr.Actors/Runtime/ActorRuntime.cs @@ -55,6 +55,7 @@ internal ActorRuntime(ActorRuntimeOptions options, ILoggerFactory loggerFactory, actor, actor.Activator ?? this.activatorFactory.CreateActivator(actor.Type), this.options.JsonSerializerOptions, + this.options.UseJsonSerialization, loggerFactory, proxyFactory, daprInteractor); diff --git a/src/Dapr.Actors/Runtime/ActorRuntimeOptions.cs b/src/Dapr.Actors/Runtime/ActorRuntimeOptions.cs index 3f4a6df88..bd68d6eb5 100644 --- a/src/Dapr.Actors/Runtime/ActorRuntimeOptions.cs +++ b/src/Dapr.Actors/Runtime/ActorRuntimeOptions.cs @@ -1,4 +1,4 @@ -// ------------------------------------------------------------------------ +// ------------------------------------------------------------------------ // Copyright 2021 The Dapr Authors // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -32,6 +32,7 @@ public sealed class ActorRuntimeOptions { Enabled = false, }; + private bool useJsonSerialization = false; private JsonSerializerOptions jsonSerializerOptions = JsonSerializerDefaults.Web; private string daprApiToken = DaprDefaults.GetDefaultDaprApiToken(); private int? remindersStoragePartitions = null; @@ -151,7 +152,22 @@ public ActorReentrancyConfig ReentrancyConfig } } - + /// + /// Enable JSON serialization for actor proxy message serialization in both remoting and non-remoting invocations. + /// + public bool UseJsonSerialization + { + get + { + return this.useJsonSerialization; + } + + set + { + this.useJsonSerialization = value; + } + } + /// /// The to use for actor state persistence and message deserialization /// diff --git a/src/Dapr.Actors/Runtime/DataContractStateSerializer.cs b/src/Dapr.Actors/Runtime/DataContractStateSerializer.cs deleted file mode 100644 index e30b21d48..000000000 --- a/src/Dapr.Actors/Runtime/DataContractStateSerializer.cs +++ /dev/null @@ -1,82 +0,0 @@ -// ------------------------------------------------------------------------ -// Copyright 2021 The Dapr Authors -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// http://www.apache.org/licenses/LICENSE-2.0 -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. -// ------------------------------------------------------------------------ - -namespace Dapr.Actors.Runtime -{ - using System; - using System.Collections.Concurrent; - using System.IO; - using System.Runtime.Serialization; - using System.Xml; - - /// - /// DataContract serializer for Actor state serialization/deserialziation. - /// This is the default state serializer used with Service Fabric Reliable Actors. - /// If there is user ask for the compatibility, this can be exposed by adding a compatibility option as an attribute on Actor type so that Service Fabric Reliable Actors state serialization behavior - /// can also be used using Dapr. - /// - internal class DataContractStateSerializer : IActorStateSerializer - { - private readonly ConcurrentDictionary actorStateSerializerCache; - - internal DataContractStateSerializer() - { - this.actorStateSerializerCache = new ConcurrentDictionary(); - } - - public byte[] Serialize(Type stateType, T state) - { - var serializer = this.actorStateSerializerCache.GetOrAdd( - stateType, - CreateDataContractSerializer); - - using var stream = new MemoryStream(); - using var writer = XmlDictionaryWriter.CreateBinaryWriter(stream); - serializer.WriteObject(writer, state); - writer.Flush(); - return stream.ToArray(); - } - - public T Deserialize(byte[] buffer) - { - if ((buffer == null) || (buffer.Length == 0)) - { - return default; - } - - var serializer = this.actorStateSerializerCache.GetOrAdd( - typeof(T), - CreateDataContractSerializer); - - using var stream = new MemoryStream(buffer); - using var reader = XmlDictionaryReader.CreateBinaryReader(stream, XmlDictionaryReaderQuotas.Max); - return (T)serializer.ReadObject(reader); - } - - private static DataContractSerializer CreateDataContractSerializer(Type actorStateType) - { - var dataContractSerializer = new DataContractSerializer( - actorStateType, - new DataContractSerializerSettings - { - MaxItemsInObjectGraph = int.MaxValue, - KnownTypes = new[] - { - typeof(ActorReference), - }, - }); - - return dataContractSerializer; - } - } -} diff --git a/test/Dapr.Actors.Test/Runtime/ActorManagerTests.cs b/test/Dapr.Actors.Test/Runtime/ActorManagerTests.cs index dd3683124..6b92c7e18 100644 --- a/test/Dapr.Actors.Test/Runtime/ActorManagerTests.cs +++ b/test/Dapr.Actors.Test/Runtime/ActorManagerTests.cs @@ -1,4 +1,4 @@ -// ------------------------------------------------------------------------ +// ------------------------------------------------------------------------ // Copyright 2021 The Dapr Authors // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -26,7 +26,7 @@ private ActorManager CreateActorManager(Type type, ActorActivator activator = nu { var registration = new ActorRegistration(ActorTypeInformation.Get(type, actorTypeName: null)); var interactor = new DaprHttpInteractor(clientHandler: null, "http://localhost:3500", apiToken: null, requestTimeout: null); - return new ActorManager(registration, activator ?? new DefaultActorActivator(), JsonSerializerDefaults.Web, NullLoggerFactory.Instance, ActorProxy.DefaultProxyFactory, interactor); + return new ActorManager(registration, activator ?? new DefaultActorActivator(), JsonSerializerDefaults.Web, false, NullLoggerFactory.Instance, ActorProxy.DefaultProxyFactory, interactor); } [Fact] diff --git a/test/Dapr.E2E.Test.Actors/ISerializationActor.cs b/test/Dapr.E2E.Test.Actors/ISerializationActor.cs new file mode 100644 index 000000000..28190a0d7 --- /dev/null +++ b/test/Dapr.E2E.Test.Actors/ISerializationActor.cs @@ -0,0 +1,22 @@ +using System.Collections.Generic; +using System.Text.Json; +using System.Text.Json.Serialization; +using System.Threading; +using System.Threading.Tasks; +using Dapr.Actors; + +namespace Dapr.E2E.Test.Actors +{ + public interface ISerializationActor : IActor, IPingActor + { + Task SendAsync(string name, SerializationPayload payload, CancellationToken cancellationToken = default); + } + + public record SerializationPayload(string Message) + { + public JsonElement Value { get; set; } + + [JsonExtensionData] + public Dictionary ExtensionData { get; set; } + } +} diff --git a/test/Dapr.E2E.Test.App/Actors/SerializationActor.cs b/test/Dapr.E2E.Test.App/Actors/SerializationActor.cs new file mode 100644 index 000000000..e8da59826 --- /dev/null +++ b/test/Dapr.E2E.Test.App/Actors/SerializationActor.cs @@ -0,0 +1,26 @@ + +using System.Threading; +using System.Threading.Tasks; +using Dapr.Actors.Runtime; + +namespace Dapr.E2E.Test.Actors.Serialization +{ + public class SerializationActor : Actor, ISerializationActor + { + public SerializationActor(ActorHost host) + : base(host) + { + } + + public Task Ping() + { + return Task.CompletedTask; + } + + public Task SendAsync(string name, + SerializationPayload payload, CancellationToken cancellationToken = default) + { + return Task.FromResult(payload); + } + } +} diff --git a/test/Dapr.E2E.Test.App/Startup.cs b/test/Dapr.E2E.Test.App/Startup.cs index b5aebbba2..34e1b2eb8 100644 --- a/test/Dapr.E2E.Test.App/Startup.cs +++ b/test/Dapr.E2E.Test.App/Startup.cs @@ -17,6 +17,7 @@ namespace Dapr.E2E.Test using Dapr.E2E.Test.Actors.Reminders; using Dapr.E2E.Test.Actors.Timers; using Dapr.E2E.Test.Actors.ExceptionTesting; + using Dapr.E2E.Test.Actors.Serialization; using Dapr.E2E.Test.App.ErrorTesting; using Dapr.Workflow; using Microsoft.AspNetCore.Authentication; @@ -34,6 +35,9 @@ namespace Dapr.E2E.Test /// public class Startup { + bool JsonSerializationEnabled => + System.Linq.Enumerable.Contains(System.Environment.GetCommandLineArgs(), "--json-serialization"); + /// /// Initializes a new instance of the class. /// @@ -83,10 +87,12 @@ public void ConfigureServices(IServiceCollection services) }); services.AddActors(options => { + options.UseJsonSerialization = JsonSerializationEnabled; options.Actors.RegisterActor(); options.Actors.RegisterActor(); options.Actors.RegisterActor(); options.Actors.RegisterActor(); + options.Actors.RegisterActor(); }); } diff --git a/test/Dapr.E2E.Test/Actors/E2ETests.CustomSerializerTests.cs b/test/Dapr.E2E.Test/Actors/E2ETests.CustomSerializerTests.cs new file mode 100644 index 000000000..c393f2ef1 --- /dev/null +++ b/test/Dapr.E2E.Test/Actors/E2ETests.CustomSerializerTests.cs @@ -0,0 +1,88 @@ +// ------------------------------------------------------------------------ +// Copyright 2021 The Dapr Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ------------------------------------------------------------------------ +namespace Dapr.E2E.Test +{ + using System; + using System.Diagnostics; + using System.Text.Json; + using System.Threading; + using System.Threading.Tasks; + using Dapr.Actors; + using Dapr.Actors.Client; + using Dapr.E2E.Test.Actors; + using Xunit; + using Xunit.Abstractions; + + public class CustomSerializerTests : DaprTestAppLifecycle + { + private readonly Lazy proxyFactory; + private IActorProxyFactory ProxyFactory => this.HttpEndpoint == null ? null : this.proxyFactory.Value; + + public CustomSerializerTests(ITestOutputHelper output, DaprTestAppFixture fixture) : base(output, fixture) + { + base.Configuration = new DaprRunConfiguration + { + UseAppPort = true, + AppId = "serializerapp", + AppJsonSerialization = true, + TargetProject = "./../../../../../test/Dapr.E2E.Test.App/Dapr.E2E.Test.App.csproj" + }; + + this.proxyFactory = new Lazy(() => + { + Debug.Assert(this.HttpEndpoint != null); + return new ActorProxyFactory(new ActorProxyOptions() { + HttpEndpoint = this.HttpEndpoint, + JsonSerializerOptions = new JsonSerializerOptions() + { + PropertyNamingPolicy = JsonNamingPolicy.CamelCase, + PropertyNameCaseInsensitive = true, + WriteIndented = true, + }, + UseJsonSerialization = true, + }); + }); + } + + [Fact] + public async Task ActorCanSupportCustomSerializer() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(60)); + var proxy = this.ProxyFactory.CreateActorProxy(ActorId.CreateRandom(), "SerializationActor"); + + await ActorRuntimeChecker.WaitForActorRuntimeAsync(this.AppId, this.Output, proxy, cts.Token); + + var payload = new SerializationPayload("hello world") + { + Value = JsonSerializer.SerializeToElement(new { foo = "bar" }), + ExtensionData = new System.Collections.Generic.Dictionary() + { + { "baz", "qux" }, + { "count", 42 }, + } + }; + + var result = await proxy.SendAsync("test", payload, CancellationToken.None); + + Assert.Equal(payload.Message, result.Message); + Assert.Equal(payload.Value.GetRawText(), result.Value.GetRawText()); + Assert.Equal(payload.ExtensionData.Count, result.ExtensionData.Count); + + foreach (var kvp in payload.ExtensionData) + { + Assert.True(result.ExtensionData.TryGetValue(kvp.Key, out var value)); + Assert.Equal(JsonSerializer.Serialize(kvp.Value), JsonSerializer.Serialize(value)); + } + } + } +} diff --git a/test/Dapr.E2E.Test/DaprRunConfiguration.cs b/test/Dapr.E2E.Test/DaprRunConfiguration.cs index 9e423205d..fccfbcdd4 100644 --- a/test/Dapr.E2E.Test/DaprRunConfiguration.cs +++ b/test/Dapr.E2E.Test/DaprRunConfiguration.cs @@ -1,4 +1,4 @@ -// ------------------------------------------------------------------------ +// ------------------------------------------------------------------------ // Copyright 2021 The Dapr Authors // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -20,9 +20,11 @@ public class DaprRunConfiguration public string AppId { get; set; } public string AppProtocol { get; set; } + + public bool AppJsonSerialization { get; set; } public string ConfigurationPath { get; set; } public string TargetProject { get; set; } } -} \ No newline at end of file +} diff --git a/test/Dapr.E2E.Test/DaprTestApp.cs b/test/Dapr.E2E.Test/DaprTestApp.cs index ee842f27b..79b4d75f5 100644 --- a/test/Dapr.E2E.Test/DaprTestApp.cs +++ b/test/Dapr.E2E.Test/DaprTestApp.cs @@ -89,6 +89,11 @@ public DaprTestApp(ITestOutputHelper output, string appId) arguments.AddRange(new[] { "--urls", $"http://localhost:{appPort.ToString(CultureInfo.InvariantCulture)}", }); } + if (configuration.AppJsonSerialization) + { + arguments.AddRange(new[] { "--json-serialization" }); + } + // TODO: we don't do any quoting right now because our paths are guaranteed not to contain spaces var daprStart = new DaprCommand(this.testOutput) {