Remote Communication mit Event-Based Components und Application Space

Wie Asynchronizität über “Zwischenstücke” in eine EBC-Architektur eingebaut werden kann, habe ich in einem früheren Blogposting beschrieben. Jetzt ist in der myCsharp.de Community die Frage aufgetaucht, wie denn eine verteilte Architektur mit EBCs realisiert werden könnte.  Die einfache Antwort: genauso ;-)

Damit meine ich, dass das, was zum Aspekt remote communication gehört, in eine EBC Standardkomponente verpackt werden sollte. Die Domänenlogik-Komponenten bekommen davon nichts mit. Hier als Beispiel eine simple Echo-Kommunikation: ein Client schickt eine Nachricht an einen Service, der sie (fast) unverändert wieder zurückschickt.

Die Kontrakte für die Komponenten sehen so aus:

public interface IClientEBC
{
    event Action<string> Out_RequestTextProcessing;

    void In_ProcessedText(string text);
}

public interface IServiceEBC
{
    void In_EchoText(string text);

    event Action<string> Out_Echo;
}

Und das sollte sich auch nicht ändern, nur weil sie nicht im selben Prozess laufen. Die Event- bzw. Nachrichtenorientierung der EBCs ist dafür eine gute Voraussetzung.

Wie die Implementationen für die Kontrakte aussieht, ist eigentlich uninteressant. Ein Blick auf die Verdrahtung lohnt allerdings. Hier der Host für beide Komponenten, solange sie im selben Prozess laufen:

namespace Servent
{
    class Program
    {
        static void Main(string[] args)
        {
            var service = new ServiceEBC();
            var client = new ClientEBC();

            client.Out_RequestTextProcessing += service.In_EchoText;
            service.Out_Echo += client.In_ProcessedText;

            Application.Run(client);
        }
    }
}

(Ich habe die Client-Komponente als WinForms-Formular ausgelegt; deshalb ruft der Host am Ende Application.Run() auf.)

Wie zu erwarten werden Instanzen beider Komponenten direkt zusammengesteckt. Der Client kommuniziert ohne Umwege mit dem Service.

Remoting zwischenstecken

Der Trick bei EBCs ist, dass Kommunikation über “Drähte” verläuft. Während Objekte normalerweise sozusagen zusammengeschweißt sind, gibt es bei EBCs immer eine Indirektion. Zwischen Client und Service sitzt ein Delegat.

Wie ich in einem früheren Blogposting gezeigt habe, ist das der Schlüssel zu großer Flexibilität. Denn wo sich Client und Service nicht wirklich “berühren”, ist es eigentlich egal, welche Distanz sie haben bzw. was zwischen ihnen sitzt.

Das nutze ich für´s Remoting nun aus. Zwischen Client und Service schiebe ich einfach zwei Standardkomponenten:

 

Die Client-Komponente kommuniziert nun nicht mehr direkt mit dem Service, sondern mit einem Proxy für ihn. Und der Service wird nicht mehr von der Client-Komponente angesprochen, sondern von einem Stub.

Ganz wichtig: Für Client und Service macht das keinen Unterschied! In WCF müssen Sie vorausschauen und für einen Service einen Service-Kontrakt definieren. Bei .NET Remoting müssen Sie einen Service durch Ableitung von MarshalByRefObject kennzeichnen. Immer, wenn Sie Funktionalität entfernt betreiben wollen, müssen Sie also speziellen Code dafür entwickeln.

Das ist hier nun anders. Der Code der Standardkomponente ist immer gleich. Sonst wäre es ja auch keine Standardkomponente ;-) Mit EBCs müssen Sie nie mehr Code für die Verteilung entwickeln. (Oder höchstens einmal für das Kommunikationsmedium Ihrer Wahl.)

Wie gesagt, Client und Service ändern sich nicht dadurch, dass Sie verteilt betrieben werden. Der Code jedoch, der die Komponenten verdrahtet, sieht anders aus. Er muss die Remoting-Infrastruktur starten und Client bzw. Service mit ihr verdrahten. Der Host für den Client sieht dann z.B. so aus:

namespace Client
{
    static class Program
    {
        [STAThread]
        static void Main()
        {
            using (var host = new RemotingHost(0)) // #1
            {
                var proxy = host.CreateProxy<string, string>(
                                     "localhost:9000/EchoService", // #2
                                     ProxyResponseHandlingModes.Sync); // #3
                var client = new ClientEBC();

                client.Out_RequestTextProcessing += proxy.In_Send;
                proxy.Out_Received += client.In_ProcessedText;

                Application.Run(client);
            } 
        }
    }
}

Der Komponenten-Host – eine Konsolenanwendung – startet die Remoting-Infrastruktur (#1) und erzeugt dann darüber einen Proxy für den Service statt des Service selbst. Wo der Service läuft, gibt eine URL an (#2). Wie die Kommunikation zwischen Proxy und Service läuft, kann dem Client egal sein. Hier kommen der Einfachheit halber aber TCP-Sockets zum Einsatz. Die Option ProxyResponseHandlingModes.Sync (#3) legt fest, dass Antworten im Synchronization Context des Client ankommen sollen; so gibt es kein Problem in WinForms-Anwendungen, denn durch das Remoting sind garantiert mehrere Threads im Spiel.

Hervorhebenswert: Der Proxy ist generisch. Er kann in jede Kommunikationsstrecke eingesetzt werden, die seinem Format folgt. Hier ist das eine bidirektionale Kommunikation auf zwei “Drähten”.

Ein Client-Host unterscheidet sich, wie Sie sehen, nur marginal von dem, der Client und Service gehostet hat. So soll es sein.

Und wie sieht es beim Service aus? Genauso simpel:

namespace Service
{
    class Program
    {
        static void Main(string[] args)
        {
            using(var host = new RemotingHost(9000)) // #1
            {
                host.CreateStub<string, string, ServiceEBC>(
                        "EchoService",
                        s => s.In_EchoText, 
                        (s, c) => s.Out_Echo += c);

                …
            } 
        }
    }
}

Der Service-Host startet dieselbe Remoting-Infrastruktur – allerdings unter Angabe eines TCP-Ports (#1). Die finden Sie beim Client in der Adresse des Service wieder (#2 im Listing davor). Der Client-Host übergibt an die Infrastruktur 0 als Port, weil ihm der Port egal ist. Er muss keinen Endpunkt definieren, weil er nicht explizit adressiert wird.

Dann erzeugt der Service-Host einen Stub für den Service. Das ist der Kommunikationsendpunkt, bei dem Nachrichten vom Proxy ankommen. Er leitet sie weiter an den eigentlichen Service. Für den Service ist der Stub der Client. Und wieder ist die Infrastruktur generisch und der eigentliche Service merkt nichts davon, dass er nun entfernt von Clients betrieben wird.

Achtung: Der Service wird hier nicht instanziert! Sein Typ und zwei Lambda Ausdrücke werden allerdings an die Factory-Methode übergeben. Die reicht sie weiter an eine ServiceFactory (s.u.). Das ermöglicht eine Erzeugung des Service für jede Nachricht wie beim Single-Call-Modus von .NET Remoting. Services zustandslos zu machen ist daher ein kleines Zugeständnis an die Verteilung. Allerdings ist das nicht zwangsläufig nötig; in diesem Fall ist es nur etwas einfacher, um die Diskussion um Remoting nicht noch mit Aspekten der asynchronen Verarbeitung zu belasten.

Das war´s. So einfach kann Remoting mit EBCs sein.

Fragt sich nur, wie es unter der Haube funktioniert :-) Ich habe es mit dem Xcoordination Application Space implementiert. Ihn habe ich in der dotnetpro beschrieben und in Grundzügen auch in meinem englischen Blog. An dieser Stelle halte ich deshalb meine Erklärungen knapp:

Host und Proxy der Remoting Infrastruktur

Der Remoting Host selbst ist einfach. Er startet eigentlich nur einen Application Space und dient als Factory für Proxy und Stub:

namespace ebc.patterns
{
    public class RemotingHost : IRemotingHost
    {
        private readonly IXcoAppSpace space;

        public RemotingHost(int tcpPort)
               : this(string.Format("tcp.port={0}", tcpPort)) { }
        public RemotingHost(string configString)
        {
            this.space = new XcoAppSpace(configString);
        }

        public void Dispose()
        {
            this.space.Dispose();
        }
    …

Ein Proxy ist schnell erzeugt:

public IRemotingProxy<TRequest, TResponse> CreateProxy<TRequest, TResponse>(
       string serviceAddress,
       ProxyResponseHandlingModes mode)
{
    var remoteStub = this.space.ConnectWorker<Port<Request<TRequest, TResponse>>>(serviceAddress);
    return new RemotingProxy<TRequest, TResponse>(remoteStub, mode);
}

Dazu nimmt der Remoting Host Kontakt mit einem generischen Service-Worker auf, dessen Adresse ihm der Client übergibt. Und der RemotingProxy übernimmt die Aufgabe, EBC-Nachrichten in AppSpace Nachrichten zu übersetzen:

namespace ebc.patterns
{
    public class RemotingProxy<TRequest, TResponse> : IRemotingProxy<TRequest, TResponse>
    {
        private readonly Port<Request<TRequest, TResponse>> remoteWorker;
        private readonly Port<TResponse> responses;

        private readonly Port<Exception> exceptions;
        private readonly SynchronizationContext ctx;

        internal RemotingProxy(Port<Request<TRequest, TResponse>> remoteWorker, ProxyResponseHandlingModes mode)
        {
            this.remoteWorker = remoteWorker;

            if (mode == ProxyResponseHandlingModes.Sync)
                this.ctx = SynchronizationContext.Current;

            // Antwort vom Service verarbeiten
            this.responses = new Port<TResponse>();
            Arbiter.Activate(
                new DispatcherQueue(),
                Arbiter.Receive(true, this.responses, msg => this.ProcessMessageInSyncContext(msg, this.Out_Received))
                );

            // Exceptions vom Service verarbeiten
            this.exceptions = new Port<Exception>();
            Arbiter.Activate(
                new DispatcherQueue(),
                Arbiter.Receive(true, this.exceptions, ex => this.ProcessMessageInSyncContext(ex, this.Out_Exception))
                );

            this.Out_Exception += ex => { };
        }

        void ProcessMessageInSyncContext<T>(T msg, Action<T> messageHandler)
        {
            if (this.ctx != null)
                this.ctx.Send(x => messageHandler(msg), null);
            else
                messageHandler(msg);
        }

        public void In_Send(TRequest message)
        {
            var c = new Causality("ex", this.exceptions);
            Dispatcher.AddCausality(c);
            {
                var req = new Request<TRequest, TResponse> {
                                  Data = message,
                                  Response = responses};
                this.remoteWorker.Post(req);
            }
            Dispatcher.RemoveCausality(c);
        }

        public event Action<TResponse> Out_Received;
        public event Action<Exception> Out_Exception;
    }
}

Außerdem übernimmt der Proxy auch die Exception-Verarbeitung. Das habe ich bisher der Einfachheit unterschlagen. Wenn Exceptions vom Service kommen, dann leitet der Proxy sie auf einem speziellen Output-Pin weiter. Damit ist die Behandlung gleich, egal ob der Client Antworten vom Service in seinem Synchronization Context empfangen will oder nicht.

Stub der Remoting Infrastruktur

Auf der Service-Seite sieht es etwas kniffliger aus. Die ist nämlich grundsätzlich multi-threaded. Derselbe Stub wird für viele Anfragen und Antworten benutzt. Daher müssen Antworten vom Service, der nichts von solchen Feinheiten weiß, mit Anfragen korreliert werden. Nur so können sie an den, der die Anfrage geschickt hat, zurückgesandt werden.  Deshalb ist der Stub nicht fest verdrahtet mit einer Service-Instanz (was allerdings möglich wäre). Das obige Bild von Stub und Service ist also eine Vereinfachung. In Wirklichkeit sind die Verhältnisse so:

Das ändert zum Glück nichts daran, dass der Service nichts von seinem Glück wissen muss, remote betrieben zu werden.

Wie funktioniert das Ganze nun? Also…

1. Eine Anfrage kommt vom RemotingProxy über ein Transportmedium im RemotingStub beim StubWorker an. Der StubWorker ist ein AppSpace Worker, der die Übersetzung von CCR Port-Kommunikation auf EBC Events übernimmt. Da er asynchron arbeitet, braucht er eine Möglichkeit, Antworten Anfragen zuzuordnen. Deshalb enhält sein Nachrichtenaustausch mit einer angeschlossenen EBC KorrelationsIDs.

namespace ebc.patterns
{
    internal class StubWorker<TRequest, TResponse>
                   : Port<Request<TRequest, TResponse>>
    {
        private readonly Dictionary<Guid, IPort> responsePorts = new Dictionary<Guid, IPort>();

        [XcoConcurrent]
        public void ProcessRequest(Request<TRequest, TResponse> req)
        {
            var msg = new CorrelatableMessage<TRequest>(req.Data);
            lock (this.responsePorts)
            {
                this.responsePorts.Add(msg.CorrelationId, req.Response);
            }
            this.Out_Received(msg);
        }

        public void In_Reply(CorrelatableMessage<TResponse> msg)
        {
            IPort response;
            lock (this.responsePorts)
            {
                response = this.responsePorts[msg.CorrelationId];
                this.responsePorts.Remove(msg.CorrelationId);
            }
            response.PostUnknownType(msg.Data);
        }

        public event Action<CorrelatableMessage<TRequest>> Out_Received;
    }
}

2. Der Kommunikationspartner des Workers kann wg. der KorrelationsIDs nicht der Service sein. Der hat von solcherlei Dingen keine Ahnung. Stattdessen reicht der Worker die Anfrage weiter an die ServiceFactory. Die erzeugt dafür eine neue Service-Instanz – und schaltet ihr eine Standardkomponente vor, die Nachrichten mit KorrelationsID in solche ohne umwandeln kann (und umgekehrt).

namespace ebc.patterns
{
    internal class ServiceFactory<TRequest, TResponse, TService>
                   : IServiceFactory<TRequest, TResponse>
        where TService : new()
    {
        private readonly Func<TService, Action<TRequest>> inputPin;
        private readonly Action<TService, Action<TResponse>> connectOutputPin;

        public ServiceFactory(
                Func<TService, Action<TRequest>> inputPin,
                Action<TService, Action<TResponse>> connectOutputPin
            )
        {
            this.inputPin = inputPin;
            this.connectOutputPin = connectOutputPin;
        }

        public void In_Request(CorrelatableMessage<TRequest> request)
        {
            var corr = new SyncCorrelator<TRequest, TResponse>();
            var service = new TService();

            corr.Out_Received += this.inputPin(service);
            this.connectOutputPin(service, corr.In_Correlate);

            corr.Out_Reply += r => this.Out_Response(r);

            corr.In_DeCorrelate(request);
        }

        public event Action<CorrelatableMessage<TResponse>> Out_Response;
    }
}

Die ServiceFactory bastelt also dynamisch für jede Anfrage einen EBC-Platineninhalt zusammen, den sie mit ihren eigenen Pins verbindet.

Das ist kein Hexenwerk, aber doch ein bisschen umständlich. Zum Glück muss es nur einmal implementiert werden und ist dann für alle Kommunikationen gleich. Einfacher wäre es allerdings, der Service wäre darauf ausgelegt, asynchron und thread-safe zu arbeiten. Dann käme er selbst mit KorrelationsIDs zurecht und müsste nicht immer neu erzeugt werden. Dann wäre das Bild tatsächlich so einfach wie oben: Stub spricht mit Service. Wie Sie sehen, geht´s aber auch so, ohne Vorüberlegungen. Das ist das schöne an EBCs.

Im folgenden Bild stelle ich die beiden alternativen mal nebeneinander:

Der Quellcode, den Sie in einem Mercurial Google Projekt hier finden, spiegel das allerdings nicht wider. Er entspricht noch dem Bild, bei dem der RemotingStub die ServiceFactory enthält. Sauberer finde ich allerdings diese letzte abgebildete Variante. Sie entzerrt die Verantwortlichkeiten. Der RemotingStub ist nur für die Kommunikation zuständig – auch wenn das bedeutet, dass er Nachrichten mit KorrelationsIDs erzeugt/konsumiert. Darauf kann sich die bei Bedarf einstellen: Ist der Service nicht damit vertraut, muss er pro Nachricht neu instanziert werden. Dazu verdrahten sie ihn passend auf einer eigenen Platine.

Fazit

Client und Service müssen nicht speziell auf eine Verteilung vorbereitet werden. Und Sie müssen für eine Verteilung auch nichts extra programmieren, keine Service- oder Datenkontrakte sind nötig. Die Remoting-Infrastruktur kann generisch sein. Sie stecken sich also zusammen, was Sie brauchen.

Das bedeutet nicht, dass Sie keinen Gedanken verschwenden sollen, ob Services lokal oder entfernt betrieben werden. Die grundsätzliche Asynchronizität (oder auch mal Unverfügbarkeit ;-) eines entfernten Dienstes können seine Implementation beeinflussen. An der grundsätzlichen Einfachheit des Zusammensteckens ändert das jedoch nichts. Mal stecken Sie nur weniger, mal mehr zusammen.

Und nun: Basteln Sie schön :-) Arbeiten Sie mit der Remoting-Infrastruktur dieses Beispiels oder bauen Sie eine eigene auf Basis von .NET Remoting oder WCF oder was Sie wollen. Bei EBCs ist das noch erlaubt, weil sie so neu sind :-)