Robotics Service Bus

using Rsb.Patterns;

            Console.WriteLine("\nRPC pattern test");

            RemoteServer remote = new RemoteServer(new Scope("/test/methods"));

            LocalServer server = new LocalServer(new Scope("/test/methods"));

            server.AddMethod("echo", new Caller());

            remote.timeout = 1000;

            //should have a timeout, since the LocalServer is not activated

            try

            {

                remote.Call<string, string>("echo", "hallo");

            }

            catch (RSBException e)

            {

                Console.WriteLine(e.Message);

            }

            server.Activate();

            // should work even though it is added after activation

            server.AddMethod("echo2", new Caller());

            Console.WriteLine("Echo : {0}", remote.Call<string, string>("echo", "hallo"));

            Console.WriteLine("Echo2: {0}", remote.Call<string, string>("echo2", "hallo"));

            Console.ReadKey();

            server.Deactivate();

    public class Caller : DataCallback<string, string>

    {

        public override string Invoke(string data)

        {

            return data + data;

        }

    }



namespace Rsb.Patterns

{

    public abstract class DataCallback<ReplyType, RequestType> : ICallback

    {

        public Event InternalInvoke(Event e) {

            if (e.Method != "REQUEST") {

                throw new RSBException("Request expected!");

            }

            ReplyType result = this.Invoke((RequestType)e.Data);

            var reply = new Rsb.Event();

            reply.Method = "REPLY";

            reply.Scope = e.Scope;

            reply.AddCause(e.EventId);

            reply.Data = result;

            return reply;

        }

        public abstract ReplyType Invoke(RequestType param);

    }

}



namespace Rsb.Patterns

{

    public interface ICallback

    {

        Event InternalInvoke(Event e);

    }

}

using System;

using System.Collections.Generic;

using System.Linq;

using System.Text;

using System.Threading.Tasks;

namespace Rsb.Patterns

{

    public class LocalServer

    {

        private Scope scope;

        private Dictionary<string, Method> registeredMethods = new Dictionary<string, Method>();

        private bool activated = false;

        // forced to be created with the Factory

        public LocalServer(Scope scope) {

            this.scope = scope;

        }

        public void AddMethod(string name, ICallback callback) {

            if (registeredMethods.ContainsKey(name)) {

                throw new ArgumentException("method already registered");

            }

            Method method = new Method(scope, name, callback);

            registeredMethods.Add(name, method);

            if (activated) {

                method.Activate();

            }

        }

        public void Activate() {

            foreach (Method method in registeredMethods.Values) {

                method.Activate();

            }

            activated = true;

        }

        public void Deactivate() {

            foreach (Method method in registeredMethods.Values) {

                method.Deactivate();

            }

            activated = false;

        }

    }

}



using System;

namespace Rsb.Patterns

{

    public class Method

    {

        private Informer informer;

        private Listener listener;

        private string name;

        public string Name { get => name; }

        private ICallback callback;

        public Method(Scope scope, string name, ICallback callback) {

            this.name = name;

            this.callback = callback;

            Scope work = scope.Concat(new Scope("/" + name));

            informer = Factory.Instance.createInformer(work);

            listener = Factory.Instance.createListener(work);

            listener.EventReceived += proceedCall;

        }

        private void proceedCall(Event theEvent)

        {

            // this is with the highest probability our own reply...

            if (theEvent.Method == "REPLY") {

                return;

            }

            // actually this would be handled in the ICallback too, but to make it more consistent...

            if (theEvent.Method != "REQUEST")

            {

                throw new RSBException("Request expected!");

            }

            informer.Send(callback.InternalInvoke(theEvent));

        }

        public void Activate() {

            informer.Activate();

            listener.Activate();

        }

        public void Deactivate()

        {

            informer.Deactivate();

            listener.Deactivate();

        }

    }

}

using System;

using System.Threading;

namespace Rsb.Patterns

{

    public class RemoteServer

    {

        public int timeout = 3000; // milliseconds

        private Scope scope;

        public RemoteServer(Scope scope) {

            this.scope = scope;

        }

        public Event Call(string name, Event request) {

            Scope methodScope = scope.Concat(new Scope("/" + name));

            // checks if the request event is well-formed instead of well-forming the event

            if (!request.Method.Equals("REQUEST")) {

                throw new RSBException(String.Format("Request event expected! (Received: {0})", request.Method));

            }

            if (!request.Scope.Equals(methodScope)) {

                throw new RSBException(String.Format("Event is not the expected scope. (Expected: {0} | Got: {1})", methodScope, request.Scope));

            }

            // creating the callback for the reply

            Event result = new Event();

            object syncPrimitive = new object();

            NewEventHandler handler = (e) => {

                if (e.Method == "REPLY" && e.IsCause(request.EventId))

                {

                    result = e;

                    // unlocking the SYNCHRONOUS call

                    lock (syncPrimitive)

                    {

                        Monitor.Pulse(syncPrimitive);

                    }

                }

            };

            Informer requestInformer = Factory.Instance.createInformer(methodScope);

            requestInformer.Activate();

            Listener replyListener = Factory.Instance.createListener(methodScope);

            replyListener.EventReceived += handler;

            replyListener.Activate();

            bool timeout;

            // Locking until reply arrives (see handler above)

            lock (syncPrimitive)

            {

                requestInformer.Send(request);

                requestInformer.Deactivate();

                timeout = !Monitor.Wait(syncPrimitive, this.timeout);

            }

            replyListener.Deactivate();

            if (timeout) {

                throw new RSBException(String.Format("Timeout after {0} milliseconds", this.timeout));

            }

            return result;

        }

        public ReplyType Call<ReplyType, RequestType>(string name, RequestType requestData) {

            Event request = new Event();

            request.Data = requestData;

            request.Scope = scope.Concat(new Scope("/" + name));

            request.Method = "REQUEST";

            Event reply = this.Call(name, request);

            if (reply.Data.GetType().Equals(typeof(ReplyType)))

            {

                return (ReplyType)reply.Data;

            }

            else {

                throw new RSBException(String.Format("Unexpected Datatype! (Expected: {0} | Got: {1})", typeof(ReplyType).ToString(), reply.Data.GetType().ToString()));

            }   

        }

    }

}

    <Compile Include="Rsb\Patterns\DataCallback.cs" />

    <Compile Include="Rsb\Patterns\ICallback.cs" />

    <Compile Include="Rsb\Patterns\LocalServer.cs" />

    <Compile Include="Rsb\Patterns\Method.cs" />

    <Compile Include="Rsb\Patterns\RemoteServer.cs" />