Пишем свой синхронный/асинхронный клиент-сервер

Всем привет.

В этой статье рассмотрим принцип многопоточного TCP сервера приложений в котором реализуем синхронные и асинхронные вызовы, а также разграничение доступа к процедурам и сжатие данных.

С чего все начиналось.
Все началось с непростого выбора с чего начать реализацию взаимодействия между собой нескольких разных приложений работающих через сеть интернет. Казалось-бы WCF способен решить такую задачу, но, к сожалению, он не лишен минусов и некоторых проблем, а его принцип работы сильно сказывается на скорости передачи данных. Нужно было простое решение и в то же время достаточно функциональное.

Начнем наш проект с библиотеки классов, в которой создадим интерфейсы процедур и функций для приложений различных типов.

public interface ICommon
{
    string[] GetAvailableUsers();
    void ChangePrivileges(string Login, string password);
}

public interface IDog
{
    bool TryFindObject(out object obj);
    int Bark(int nTimes);
}

public interface ICat
{
    void CutTheText(ref string Text);
}

Здесь мы описали сигнатуру процедур для трех разных уровней доступа. Как можно догадаться, Common будет содержать процедуры доступные для всех типов удаленных клиентов. Dog и Cat здесь, это наши два типа удаленных клиентов, процедуры каждого из них будут доступны только им самим.

Здесь же создадим класс, с помощью которого мы будем обмениваться данными между сервером и клиентами.

[Serializable]
public class Message
{
    public Message(string Command, object[] Parameters)
    {
        this.Command = Command;
        if (Parameters != null) this.prms = Parameters;
    }

    public bool IsSync;
    public bool IsEmpty = true;
    public readonly string Command;
    public object ReturnValue;
    public object[] prms;
    public Exception Exception;
}

Клиент

Клиент будет реализовать проксирующую связь между методами интерфейса и сервером. Для этого создадим класс реализующий прокси:

private class Proxy<T> : RealProxy where T : class
{
    UniservClient client;

    public Proxy(UniservClient client): base(typeof(T))
    {
        this.client = client;
    }

    public override IMessage Invoke(IMessage msg)
    {
        IMethodCallMessage call = (IMethodCallMessage)msg;
        object[] parameters = call.Args;
        int OutArgsCount = call.MethodBase.GetParameters().Where(x => x.IsOut).Count();

        Message result = client.Execute(call.MethodName, parameters);
        parameters = parameters.Select((x, index) => result.prms[index] ?? x).ToArray();
        return new ReturnMessage(result.ReturnValue, parameters, OutArgsCount, call.LogicalCallContext, call);
    }
}

И создадим свойства для доступа к методам интерфейсов:

public ICommon Common { get; private set; }
public IDog Dog { get; private set; }
public ICat Cat { get; private set; }

Инициализируем прокси и свойства:

CommonProxy = new Proxy<ICommon>(this);
DogProxy = new Proxy<IDog>(this);
CatProxy = new Proxy<ICat>(this);

Common = (ICommon)CommonProxy.GetTransparentProxy();
Dog = (IDog)DogProxy.GetTransparentProxy();
Cat = (ICat)CatProxy.GetTransparentProxy();

Обработка команд сервера:

private void Listener()
{
    while (true)
    {
        try
        {
            if (ListenerToken.IsCancellationRequested) return;

            if (!IsConnected) _Connect();

            while (true)
            {
                if (ListenerToken.IsCancellationRequested) return;

                Message msg = ReceiveData<Message>();
                if (msg.Command == "OnPing")
                {
                    // отражаем пинг
                    SendData(msg);
                    if (Events.OnPing != null) Events.OnPing.BeginInvoke(null, null);
                    continue;
                }

                if (msg.IsSync)
                {  // получен результат синхронной процедуры
                    SyncResult(msg);
                }
                else
                {
                    // асинхронный вызов события
                    try
                    {
                        // ищем соответствующий Action
                        var pi = typeof(IEvents).GetProperty(msg.Command, BindingFlags.Instance | BindingFlags.Public);
                        if (pi == null) throw new Exception(string.Concat("Свойство "", msg.Command, "" не найдено"));
                        var delegateRef = pi.GetValue(this, null) as Delegate;

                        // инициализируем событие
                        if (delegateRef != null) ThreadPool.QueueUserWorkItem(state => delegateRef.DynamicInvoke(msg.prms));
                    }
                    catch (Exception ex)
                    {
                        throw new Exception(string.Concat("Не удалось выполнить делегат "", msg.Command, """), ex);
                    }
                }
            }
        }
        catch (TaskCanceledException)
        {
            return;
        }
        catch (Exception ex)
        {
            if (Events.OnError != null) Events.OnError.BeginInvoke(ex, null, null);
        }
        finally
        {
            _Dicsonnect();
        }

        Thread.Sleep(2000);
    }
}

За выполнение удаленных процедуры отвечают методы:

private Message Execute(string MethodName, object[] parameters)
{
    lock (syncLock)
    {
        _syncResult = new Message(MethodName, parameters);
        _syncResult.IsSync = true;

        _OnResponce.Reset(); 
        SendData(_syncResult);
        _OnResponce.Wait();  // ожидаем ответ сервера

        if (_syncResult.IsEmpty)
        {// произошел дисконект, результат не получен
            throw new Exception(string.Concat("Ошибка при получении результата на команду "", MethodName, """));
        }

        if (_syncResult.Exception != null) throw _syncResult.Exception;  // исключение переданное сервером
        return _syncResult;
    }
}

private void SyncResult(Message msg)
{  // получен результат выполнения процедуры

    _syncResult = msg;
    _syncResult.IsEmpty = false;

    _OnResponce.Set();  // разблокируем поток
}

В общих чертах клиент работает следующим образом:
Когда приложение вызывает метод интерфейса, выполнение переходит к прокси-серверу. Прокси отправляет серверу имя вызываемого метода вместе с его параметрами и блокирует поток в ожидании результата выполнения. Разбор ответов сервера происходит в другом потоке, который разрешает продолжение заблокированного потока.

Сервер

Создадим иерархию классов, которые будут определять доступность тех или иных функций:

public class Cat_Ring0 : Ring2, ICat
{
    public Cat_Ring0(User u) : base(u)
    {
        up.UserType = UserType.Cat;
    }

    public void CutTheText(ref string Text)
    {
        Text = Text.Remove(Text.Length - 1);
    }
}

public class Dog_Ring0 : Dog_Ring1, IDog
{
    public Dog_Ring0(User u) : base(u)
    {
        up.UserType = UserType.Dog;
    }

    public int Bark(int nTimes)
    {
        var ConnectedDogs = ConnectedUsers.ToArray().Where(x => x.UserType == UserType.Dog).Select(x => x.nStream);
        ConnectedDogs.AsParallel().ForAll(nStream =>
        {
            SendMessage(nStream, new Message("OnBark", new object[] { nTimes}));
        });

        return ConnectedDogs.Count();
    }
}

public class Dog_Ring1 : Ring2
{
    public Dog_Ring1(User u): base(u)
    {
        up.UserType = UserType.Dog;
    }

    public bool TryFindObject(out object obj)
    {
        obj = "TheBall";
        return true;
    }
}

public class Ring2 : Ring, ICommon
{
    public Ring2(User u) : base(u) { }

    public string[] GetAvailableUsers()
    {
        return new string[] { "Dog0", "Dog1", "Tom" };
    }

    public void ChangePrivileges(string Animal, string password)
    {
        switch (Animal)
        {
            case "Dog0":
                if (password != "groovy!") throw new Exception("Не верный пароль");
                up.ClassInstance = new Dog_Ring0(up);
                break;
            case "Dog1":
                if (password != "_password") throw new Exception("Не верный пароль");
                up.ClassInstance = new Dog_Ring1(up);
                break;
            case "Tom":
                if (password != "TheCat") throw new Exception("Не верный пароль");
                up.ClassInstance = new Cat_Ring0(up);
                break;
            default:
                throw new Exception("Такого пользователя не существует");
        }
    }
}

public abstract class Ring
{
    public readonly User up;

    public Ring(User up)
    {
        this.up = up;
    }
}

Теперь достаточно поместить процедуру в определенное “кольцо” что бы соответствующий клиент имел к ней доступ. Ring0 это верхний уровень доступа, пользователь этого типа имеет доступ не только к находящимся в нем процедурам, но и процедурам во всех наследуемых классах. Изначально пользователь попадает в Ring2, который реализует только общие методы доступные всем. Далее с помощью ChangePrivileges() пользователь может, пройдя авторизацию, попасть в определенный тип «кольца» с определенным уровнем доступа.

Основная работа сервера сводится к следующему методу:

private void ProcessMessage(Message msg, User u)
{
    string MethodName = msg.Command;
    if (MethodName == "OnPing") return;

    // ищем запрошенный метод в кольце текущего уровня
    MethodInfo method = u.RingType.GetMethod(MethodName, BindingFlags.Instance | BindingFlags.Public | BindingFlags.FlattenHierarchy);

    try
    {
        if (method == null)
        {
            throw new Exception(string.Concat("Метод "", MethodName, "" недоступен"));
        }

        try
        {
            // выполняем метод интерфейса
            msg.ReturnValue = method.Invoke(u.ClassInstance, msg.prms);
        }
        catch (Exception ex)
        {
            throw ex.InnerException;
        }

        // возвращаем ref и out параметры
        msg.prms = method.GetParameters().Select(x => x.ParameterType.IsByRef ? msg.prms[x.Position] : null).ToArray();
    }
    catch (Exception ex)
    {
        msg.Exception = ex;
    }
    finally
    {
        // возвращаем результат выполнения запроса
        SendMessage(u.nStream, msg);
    }
}

Свойство ClassInstance содержит экземпляр “кольца” в котором будет выполняться поиск процедуры по ее имени.

Пример лога выполнения:

В результате получилось простое и элегантное решение аналогичное WCF.
Исходник можно взять тут ^[1]

Автор: vitaliy91

Источник ^[2]