Robotics Service Bus

using Rsb.Patterns;

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

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

            remote.Activate();

            remote.timeout = 1000;

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

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

            //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();

            for (int i = 0; i < 20; i++) {

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

            }

            Console.ReadKey();

            server.Deactivate();

            remote.Deactivate();

	}

    public class Caller : DataCallback<string, string>

    {

        public override string Invoke(string data)

        {

            return data + data;

        }

        public RemoteServer CreateRemoteServer(Scope scope, int timeout = 3000)

        {

            var server = new RemoteServer(scope)

            {

                Timeout = timeout

            };

            return server;

        }

        public RemoteServer CreateRemoteServer(string scope, int timeout = 3000)

        {

            return CreateRemoteServer(new Scope(scope), timeout);

        }

        public LocalServer CreateLocalServer(Scope scope)

        {

            var server = new LocalServer(scope);

            return server;

        }

        public LocalServer CreateLocalServer(string scope)

        {

            return CreateLocalServer(new Scope(scope));

        }



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>();

        public bool IsActive {

            get; private set;

        }

        public LocalServer(Scope scope) {

            this.scope = scope;

            IsActive = false;

        }

        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 (IsActive) {

                method.Activate();

            }

        }

        public void Activate() {

            foreach (Method method in registeredMethods.Values) {

                method.Activate();

            }

            IsActive = true;

        }

        public void Deactivate() {

            foreach (Method method in registeredMethods.Values) {

                method.Deactivate();

            }

            IsActive = 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;

using Rsb;

namespace Rsb.Patterns

{

    public class RemoteMethod

    {

        Informer requestInformer;

        Listener replyListener;

        int timeout;

        object syncPrimitive;

        EventId requestId;

        Event result;

        public bool IsActive { get; private set; }

        public RemoteMethod(Scope scope, int timeout) {

            requestInformer = Factory.Instance.createInformer(scope);

            replyListener = Factory.Instance.createListener(scope);

            replyListener.EventReceived += handler;

            this.timeout = timeout;

            syncPrimitive = new object();

            IsActive = false;

        }

        private void handler(Event e) {

            if (requestId == null) {

                //maybe exception but actually we can just ignore...

                return;

            }

            if (e.Method == "REPLY" && e.IsCause(requestId))

            {

                // unlocking the SYNCHRONOUS call

                lock (syncPrimitive)

                {

                    result = e;

                    requestId = null;

                    Monitor.Pulse(syncPrimitive);

                }

            }

        }

        public Event Call(Event request) {

            if (!IsActive) {

                throw new InvalidOperationException("Participant not active");

            }

            bool timeout;

            // Locking until reply arrives (see handler above)

            lock (syncPrimitive)

            {

                requestId = request.EventId;

                requestInformer.Send(request);

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

            }

            if (timeout)

            {

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

            }

            return result;

        }

        public void Activate() {

            requestInformer.Activate();

            replyListener.Activate();

            IsActive = true;

        }

        public void Deactivate() {

            requestInformer.Deactivate();

            replyListener.Deactivate();

            IsActive = false;

        }

    }

}

using System;

using System.Collections.Generic;

using System.Threading;

namespace Rsb.Patterns

{

    public class RemoteServer

    {

        public int Timeout = 3000; // milliseconds

        private Scope scope;

        public bool IsActive { get; private set; }

        Dictionary<string, RemoteMethod> connections = new Dictionary<string, RemoteMethod>();

        public RemoteServer(Scope scope) {

            this.scope = scope;

            IsActive = false;

        }

        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));

            }

            RemoteMethod method;

            if (connections.ContainsKey(name)) {

                method = connections[name];

            }

            else

            {

                method = new RemoteMethod(methodScope, Timeout);

                connections.Add(name, method);

                if (IsActive) {

                    method.Activate();

                }

            }

            return method.Call(request);

        }

        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()));

            }   

        }

        public void Activate() {

            foreach (var remoteMethod in connections.Values) {

                remoteMethod.Activate();

            }

            IsActive = true;

        }

        public void Deactivate() {

            foreach (var remoteMethod in connections.Values) {

                remoteMethod.Deactivate();

            }

            IsActive = false;

        }

    }

}

    <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\RemoteMethod.cs" />

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