Trouble Connecting to Custom Akka.NET Transport

Question

I am trying to implement a custom Akka.NET Transport. Currently I have the following:

Client program (based on https://getakka.net/articles/remoting/messaging.html):

using Akka.Actor;
using Akka.Configuration;
using AkkaTest.Messages;

namespace AkkaTest;

static class Program
{
    static void Main()
    {
        var config = ConfigurationFactory.ParseString(@"
akka {  
    actor {
        provider = remote
    }
    remote {
        enabled-transports = [""akka.remote.myproto""]
        myproto {
            transport-class = ""AkkaTest.Messages.MyTransport, AkkaTest.Messages""
            applied-adapters = []
            transport-protocol = myproto
            port = 0 # bound to a dynamic port assigned by the OS
            hostname = localhost
        }
    }
}");

        var system = ActorSystem.Create("MyActorSystem", config);
        var myActor = system.ActorOf<SendActor>();
        myActor.Tell(new StartRemoting());
        while (true)
        {
            // null task
        }
    }
}

// Initiates remote contact 
public class SendActor : ReceiveActor
{
    public SendActor()
    {
        Receive<StartRemoting>(s => {
            Context.ActorSelection("akka.myproto://RemoteSys@localhost:8081/user/Echo")
            .Tell(new Msg("hi!"));
        });
        Receive<Msg>(msg => {
            Console.WriteLine("Received {0} from {1}", msg.Content, Sender);
        });
    }
}

Server:

using Akka.Actor;
using Akka.Configuration;
using AkkaTest.Messages;

namespace AkkaTest.Server;

static class Program
{
    static void Main()
    {
        var config = ConfigurationFactory.ParseString(@"
akka {  
    actor {
        provider = remote
    }
    remote {
        enabled-transports = [""akka.remote.myproto""]
        myproto {
            transport-class = ""AkkaTest.Messages.MyTransport, AkkaTest.Messages""
            applied-adapters = []
            transport-protocol = myproto
            port = 8081 #bound to a specific port
            hostname = localhost
        }
    }
}");

        var system = ActorSystem.Create("RemoteSys", config);
        system.ActorOf<EchoActor>("Echo");
        while (true)
        {
            // null task
        }
    }
}

// Runs in a separate process from SendActor
public class EchoActor : ReceiveActor
{
    public EchoActor()
    {
        Receive<Msg>(msg => {
            // echo message back to sender
            Sender.Tell(msg);
        });
    }
}

Messages:

namespace AkkaTest.Messages;

// Written into a shared library
public class Msg
{
    public Msg(string content)
    {
        this.Content = content;
    }

    public string Content { get; set; }
}

public class StartRemoting { }

Stubbed-out association event listener:

namespace AkkaTest.Messages
{
    using Akka.Remote.Transport;

    public class MyAel : IAssociationEventListener
    {
        public void Notify(IAssociationEvent ev)
        {
            throw new NotImplementedException();
        }
    }
}

And finally, the stubbed-out transport:

namespace AkkaTest.Messages
{
    using Akka.Actor;
    using Akka.Configuration;
    using Akka.Remote.Transport;
    using System;
    using System.Threading.Tasks;

    public class MyTransport : Transport
    {
        public MyTransport(ActorSystem system, Config config)
        {
            this.System = system;
            this.Config = config;
        }

        public override Task<AssociationHandle> Associate(Address remoteAddress)
        {
            throw new NotImplementedException();
        }

        public override bool IsResponsibleFor(Address remote)
        {
            return remote.Protocol.ToUpper() == this.Config.GetString("transport-protocol").ToUpper();
        }

        public override Task<(Address, TaskCompletionSource<IAssociationEventListener>)> Listen()
        {
            var tcs = new TaskCompletionSource<IAssociationEventListener>();
            tcs.TrySetResult(new MyAel());
            var protocol = this.Config.GetString("transport-protocol");
            var host = this.Config.GetString("hostname");
            var port = this.Config.GetInt("port");
            if (port == 0)
            {
                port = 62978;
            }
            return Task.FromResult((
                new Address(
                    protocol,
                    this.System.Name,
                    host,
                    port),
                tcs));
        }

        public override Task<bool> Shutdown()
        {
            throw new NotImplementedException();
        }
    }
}

Previously, I had tried a very similar client/server setup with TCP instead of my custom transport. This was the server output:

[INFO][9/27/2022 2:29:15 PM][Thread 0001][remoting (akka://RemoteSys)] Starting remoting
[INFO][9/27/2022 2:29:15 PM][Thread 0001][remoting (akka://RemoteSys)] Remoting started; listening on addresses : [akka.tcp://RemoteSys@localhost:8081]
[INFO][9/27/2022 2:29:15 PM][Thread 0001][remoting (akka://RemoteSys)] Remoting now listens on addresses: [akka.tcp://RemoteSys@localhost:8081]

And the client:

[INFO][9/27/2022 2:29:29 PM][Thread 0001][remoting (akka://MyActorSystem)] Starting remoting
[INFO][9/27/2022 2:29:30 PM][Thread 0001][remoting (akka://MyActorSystem)] Remoting started; listening on addresses : [akka.tcp://MyActorSystem@localhost:62978]
[INFO][9/27/2022 2:29:30 PM][Thread 0001][remoting (akka://MyActorSystem)] Remoting now listens on addresses: [akka.tcp://MyActorSystem@localhost:62978]
Received hi! from [akka.tcp://RemoteSys@localhost:8081/user/Echo#2122916400]

So this is my expected behavior. When I start the server project with my custom transport, I get the following console output:

[INFO][9/27/2022 3:49:50 PM][Thread 0001][remoting (akka://RemoteSys)] Starting remoting
[INFO][9/27/2022 3:49:50 PM][Thread 0001][remoting (akka://RemoteSys)] Remoting started; listening on addresses : [akka.myproto://RemoteSys@localhost:8081]
[INFO][9/27/2022 3:49:50 PM][Thread 0001][remoting (akka://RemoteSys)] Remoting now listens on addresses: [akka.myproto://RemoteSys@localhost:8081]

This is what I would expect. Very similar to the output I got when trying the same setup with TCP. However, when I try starting the client, I get the following error:

[INFO][9/27/2022 3:50:48 PM][Thread 0001][remoting (akka://MyActorSystem)] Starting remoting
[INFO][9/27/2022 3:50:48 PM][Thread 0001][remoting (akka://MyActorSystem)] Remoting started; listening on addresses : [akka.myproto://MyActorSystem@localhost:62978]
[INFO][9/27/2022 3:50:48 PM][Thread 0001][remoting (akka://MyActorSystem)] Remoting now listens on addresses: [akka.myproto://MyActorSystem@localhost:62978]
[ERROR][9/27/2022 3:50:49 PM][Thread 0009][akka://MyActorSystem/user/$a] No transport is loaded for protocol: [akka.myproto], available protocols: [akka.]
Cause: Akka.Remote.RemoteTransportException: No transport is loaded for protocol: [akka.myproto], available protocols: [akka.]
   at Akka.Remote.Remoting.LocalAddressForRemote(IDictionary`2 transportMapping, Address remote)
   at Akka.Remote.RemoteActorRefProvider.RootGuardianAt(Address address)
   at Akka.Actor.ActorRefFactoryShared.ActorSelection(String path, ActorSystem system, IActorRef lookupRoot)
   at AkkaTest.SendActor.<>c.<.ctor>b__0_0(StartRemoting s) in ...\AkkaTest\AkkaTest\Program.cs:line 44
   at lambda_method7(Closure , Object , Action`1 , Action`1 )
   at Akka.Actor.ReceiveActor.OnReceive(Object message)
   at Akka.Actor.UntypedActor.Receive(Object message)
   at Akka.Actor.ActorBase.AroundReceive(Receive receive, Object message)
   at Akka.Actor.ActorCell.ReceiveMessage(Object message)
   at Akka.Actor.ActorCell.Invoke(Envelope envelope)

I get the same issue regardless of whether I have the server started or not, so it's almost as if it's not even trying to make the association. I suspect the problem might be that I just have too much stubbed out at this point (I'm particularly wondering if the issue might be with the tcs.TrySetResult(new MyAel()); line), but I don't have a lot to go on for what to fill in; the documentation is a bit sparse. So far as I know, none of the NotImplementedExceptions are being hit, nor any preceding breakpoints. So what am I missing at this stage?

Answer 1

Found the (current) issue. Examining the metadata for DotnettyTransport, I noticed it overrode the virtual SchemeIdentifier. I set the SchemeIdentifier in my ctor:

public MyTransport(ActorSystem system, Config config)
{
    this.System = system;
    this.Config = config;
    this.SchemeIdentifier = this.Config.GetString("transport-protocol");
}

After I did this, my IsResponsibleFor implementation got called. That led to my next issue, the incoming remote.Protocol property did not return myproto like I expected, but akka.myproto. Easy enough to fix:

public override bool IsResponsibleFor(Address remote)
{
    return remote.Protocol.ToUpper() == $"AKKA.{this.SchemeIdentifier.ToUpper()}";
}

Might tidy that up later. Now that that is working, my Associate override is getting called, and of course throwing its NotImplemented[yet]Exception.