Akka.net: Issue with StreamsRef over remote system

[pm-mw]

I have build a small test case for the beta StreamsRef feature (1.4 branch), very similar to the unit test, which is used in the docs for StreamsRef but using two (remote) Akka systems.

The relevant code is here https://gist.github.com/pm-mw/6df9e955412ae5c1087ad10e741d166e
When the source actor receives the Receive message and pipes the stream to the sink (receiving actor) I get the exception

[ERROR][25.04.2019 06:45:15][Thread 0011][remoting] Die angegebene Methode wird nicht unterstützt.
Cause: System.NotSupportedException: Die angegebene Methode wird nicht unterstützt.
   bei System.Collections.Immutable.ImmutableArray`1.System.Collections.Generic.ICollection<T>.Add(T item)
   bei Newtonsoft.Json.Utilities.CollectionWrapper`1.Add(T item)
   bei Newtonsoft.Json.Utilities.CollectionWrapper`1.System.Collections.IList.Add(Object value)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.CreateObjectUsingCreatorWithParameters(JsonReader reader, JsonObjectContract contract, JsonProperty containerProperty, ObjectConstructor`1 creator, String id)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.CreateNewObject(JsonReader reader, JsonObjectContract objectContract, JsonProperty containerMember, JsonProperty containerProperty, String id, Boolean& createdFromNonDefaultCreator)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.CreateObject(JsonReader reader, Type objectType, JsonContract contract, JsonProperty member, JsonContainerContract containerContract, JsonProperty containerMember, Object existingValue)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.CreateValueInternal(JsonReader reader, Type objectType, JsonContract contract, JsonProperty member, JsonContainerContract containerContract, JsonProperty containerMember, Object existingValue)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.ResolvePropertyAndCreatorValues(JsonObjectContract contract, JsonProperty containerProperty, JsonReader reader, Type objectType)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.CreateObjectUsingCreatorWithParameters(JsonReader reader, JsonObjectContract contract, JsonProperty containerProperty, ObjectConstructor`1 creator, String id)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.CreateNewObject(JsonReader reader, JsonObjectContract objectContract, JsonProperty containerMember, JsonProperty containerProperty, String id, Boolean& createdFromNonDefaultCreator)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.CreateObject(JsonReader reader, Type objectType, JsonContract contract, JsonProperty member, JsonContainerContract containerContract, JsonProperty containerMember, Object existingValue)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.CreateValueInternal(JsonReader reader, Type objectType, JsonContract contract, JsonProperty member, JsonContainerContract containerContract, JsonProperty containerMember, Object existingValue)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.ResolvePropertyAndCreatorValues(JsonObjectContract contract, JsonProperty containerProperty, JsonReader reader, Type objectType)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.CreateObjectUsingCreatorWithParameters(JsonReader reader, JsonObjectContract contract, JsonProperty containerProperty, ObjectConstructor`1 creator, String id)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.CreateNewObject(JsonReader reader, JsonObjectContract objectContract, JsonProperty containerMember, JsonProperty containerProperty, String id, Boolean& createdFromNonDefaultCreator)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.CreateObject(JsonReader reader, Type objectType, JsonContract contract, JsonProperty member, JsonContainerContract containerContract, JsonProperty containerMember, Object existingValue)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.CreateValueInternal(JsonReader reader, Type objectType, JsonContract contract, JsonProperty member, JsonContainerContract containerContract, JsonProperty containerMember, Object existingValue)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.ResolvePropertyAndCreatorValues(JsonObjectContract contract, JsonProperty containerProperty, JsonReader reader, Type objectType)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.CreateObjectUsingCreatorWithParameters(JsonReader reader, JsonObjectContract contract, JsonProperty containerProperty, ObjectConstructor`1 creator, String id)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.CreateNewObject(JsonReader reader, JsonObjectContract objectContract, JsonProperty containerMember, JsonProperty containerProperty, String id, Boolean& createdFromNonDefaultCreator)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.CreateObject(JsonReader reader, Type objectType, JsonContract contract, JsonProperty member, JsonContainerContract containerContract, JsonProperty containerMember, Object existingValue)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.CreateValueInternal(JsonReader reader, Type objectType, JsonContract contract, JsonProperty member, JsonContainerContract containerContract, JsonProperty containerMember, Object existingValue)
   bei Newtonsoft.Json.Serialization.JsonSerializerInternalReader.Deserialize(JsonReader reader, Type objectType, Boolean checkAdditionalContent)
   bei Newtonsoft.Json.JsonSerializer.DeserializeInternal(JsonReader reader, Type objectType)
   bei Newtonsoft.Json.JsonConvert.DeserializeObject(String value, Type type, JsonSerializerSettings settings)
   bei Akka.Serialization.NewtonSoftJsonSerializer.FromBinary(Byte[] bytes, Type type)
   bei Akka.Serialization.Serialization.Deserialize(Byte[] bytes, Int32 serializerId, String manifest)
   bei Akka.Remote.MessageSerializer.Deserialize(ActorSystem system, Payload messageProtocol)
   bei Akka.Remote.DefaultMessageDispatcher.Dispatch(IInternalActorRef recipient, Address recipientAddress, Payload message, IActorRef senderOption)
   bei Akka.Remote.EndpointReader.<Reading>b__11_1(InboundPayload inbound)
   bei lambda_method(Closure , Object , Action`1 , Action`1 , Action`1 )
   bei Akka.Tools.MatchHandler.PartialHandlerArgumentsCapture`4.Handle(T value)
   bei Akka.Actor.ReceiveActor.ExecutePartialMessageHandler(Object message, PartialAction`1 partialAction)
   bei Akka.Actor.ReceiveActor.OnReceive(Object message)
   bei Akka.Actor.UntypedActor.Receive(Object message)
   bei Akka.Actor.ActorBase.AroundReceive(Receive receive, Object message)
   bei Akka.Actor.ActorCell.ReceiveMessage(Object message)
   bei Akka.Actor.ActorCell.Invoke(Envelope envelope)

(Method not supported).

As far as I can see all messages seem to be serializable. Or is sth wrong with my appraoch ? I can also upload the full project if needed. Thanks.

[valdisz]

I will look on to this issue

[valdisz]

@pm-mw it would be nice to get the whole solution, otherwise, it is hard to reproduce. Can you provide it?

[pm-mw]

https://github.com/pm-mw/DemoAkkaStreamsRemote

Thanks!

[valdisz]

@pm-mw tnx for code, even after an update to latest Akka version (beta2) it still fails. I will dig into what happens.

[valdisz]

What I have found so far is that the issue is related to how Newotnsoft.JSON is used. My first assumption was that it is bug in the JSON library, but upgrading to the latest version still crashes just with a different exception.

[[akka://system2/system/endpointManager/reliableEndpointWriter-akka.tcp%3A%2F%2Fsystem1%40localhost%3A8090-1/endpointWriter#648718772]] AssociationError [akka.tcp://system2@localhost:5156] -> akka.tcp://system1@localhost:8090: Error [Cannot preserve reference to readonly dictionary, or dictionary created from a non-default constructor: System.Collections.Immutable.IImmutableDictionary`2[Akka.Streams.InPort,Akka.Streams.OutPort]. Path 'SourceRef.Source.Module.Upstreams', line 1, position 1197.] [ 
   at Akka.Remote.EndpointReader.<Reading>b__11_1(InboundPayload inbound) in C:\local\var\git-private\akka.net\src\core\Akka.Remote\Endpoint.cs:line 1938
   at lambda_method(Closure , Object , Action`1 , Action`1 , Action`1 )
   at Akka.Tools.MatchHandler.PartialHandlerArgumentsCapture`4.Handle(T value) in C:\local\var\git-private\akka.net\src\core\Akka\Util\MatchHandler\PartialHandlerArgumentsCapture.cs:line 143
   at Akka.Actor.ReceiveActor.ExecutePartialMessageHandler(Object message, PartialAction`1 partialAction) in C:\local\var\git-private\akka.net\src\core\Akka\Actor\ReceiveActor.cs:line 78
   at Akka.Actor.ReceiveActor.OnReceive(Object message) in C:\local\var\git-private\akka.net\src\core\Akka\Actor\ReceiveActor.cs:line 73
   at Akka.Actor.UntypedActor.Receive(Object message) in C:\local\var\git-private\akka.net\src\core\Akka\Actor\UntypedActor.cs:line 26
   at Akka.Actor.ActorBase.AroundReceive(Receive receive, Object message) in C:\local\var\git-private\akka.net\src\core\Akka\Actor\ActorBase.cs:line 158
   at Akka.Actor.ActorCell.ReceiveMessage(Object message) in C:\local\var\git-private\akka.net\src\core\Akka\Actor\ActorCell.DefaultMessages.cs:line 177
   at Akka.Actor.ActorCell.Invoke(Envelope envelope) in C:\local\var\git-private\akka.net\src\core\Akka\Actor\ActorCell.DefaultMessages.cs:line 81]

[valdisz]

@Aaronontheweb I think this is bug

[Aaronontheweb]

@valdisz @pm-mw yeah, that would do it - means that the custom serialization we use for StreamRefs isn't being used and it's falling back to Newtonsoft.Json serialization, which doesn't support any of the System.Collections.Immutable data structures:

https://github.com/akkadotnet/akka.net/blob/dev/src/core/Akka.Streams/Serialization/StreamRefSerializer.cs

https://github.com/akkadotnet/akka.net/blob/07990753bdf87e9d1ec1e605bb01affaedd5c027/src/core/Akka.Streams/reference.conf#L143-L158

I don't think we have any tests in the suite right now that explicitly check to see if this serializer is being used for when StreamRefs are passed over the network, not in this class at least: https://github.com/akkadotnet/akka.net/blob/dev/src/core/Akka.Streams.Tests/Dsl/StreamRefsSpec.cs

@valdisz could you add one? That would give us a clear idea if the issue is with the akka.actor.serialization HOCON not being loaded or written properly for StreamRef support. There might already be one on the JVM too.

[valdisz]

Yes, sure... just give me time for that :)

[valdisz]

@pm-mw @Aaronontheweb my current understanding is that the problem is in the following: IStreamRef is placed into envelope what is essentially a POCO class. Because serializer is selected based on message type (which is POCO), but not properties, we have failure here.

If IStreamRef is sent directly as a message, then everything works as expected.

This is by design of current serializers. To allow enveloped IStreamRef some design changes must be made what is not in my competence.

I can imagine that we can make a special marker interface for such a case and handle this case in StreamRefSerializer with appropriate type bindings.

All in all, if @Aaronontheweb will point to direction how he wants to see this implemented, I can make it :)

[Aaronontheweb]

IStreamRef is placed into envelope what is essentially a POCO class. Because serializer is selected based on message type (which is POCO), but not properties, we have failure here.

Ah yes, that would do it... I see what you mean.

Might need to do something similar to what we do with IActorRef serialization - we know how to convert it back and forth even when it's included on messages. I wonder if this has come up at all on the JVM before? I'd research that first since they've had stream refs in production for longer than we have.

[valdisz]

@Aaronontheweb by implementing ISurrogated all is serialized nicely I will make implementation to look good tomorrow then you can review it :)

[Aaronontheweb]

Nice work - just need API approval and we can get this change added to v1.4.0-beta4.

[Aaronontheweb]

closed via https://github.com/akkadotnet/akka.net/pull/4063