Перейти к содержанию

Отправка и получение сообщений Protobuf

Процесс отправки и получения сообщения различается для стандартов TCP и WebSocket. Ниже мы подробно объясняем этот процесс для обоих типов соединений.

Использование TCP

Отправка сообщения

Для отправки сообщения через соединение TCP выполните следующие действия:

  1. Преобразуйте сообщение Protobuf в массив байтов (используя кодировку Protobuf) с помощью официального SDK Google Protocol Buffer для выбранного вами языка программирования.

  2. Получите длину массива, созданного на шаге 1. Создайте новый массив байтов из этого целого числа. Переверните новый массив байтов.

  3. Объедините новый массив байтов и массив байтов, содержащий исходное сообщение Protobuf.

  4. Отправьте объединенный массив в поток соединения.

Приведенные ниже примеры демонстрируют, как эти шаги выполняются в официальных SDK Open API.

1
2
3
4
5
6
7
private async Task WriteTcp(byte[] messageByte, CancellationToken cancellationToken)
{
    byte[] array = BitConverter.GetBytes(messageByte.Length).Reverse().Concat(messageByte)
        .ToArray();
    await _sslStream.WriteAsync(array, 0, array.Length, cancellationToken).ConfigureAwait(continueOnCapturedContext: false);
    await _sslStream.FlushAsync(cancellationToken).ConfigureAwait(continueOnCapturedContext: false);
}

1
2
3
client = Client(EndPoints.PROTOBUF_LIVE_HOST if hostType.lower() == "live" else EndPoints.PROTOBUF_DEMO_HOST, EndPoints.PROTOBUF_PORT, TcpProtocol)
request = ProtoOAApplicationAuthReq() # Can be any message
deferred = client.send(request)

Пример на Python

Используя Twisted, пример на Python выполняет практически те же операции, что и пример на C#. Команду client.send(request) можно объяснить следующим образом.

1
2
3
request = ProtoOAApplicationAuthReq() # Can be any message
requestAsString = request.SerializeToString() # This method is a part of the Google Protobuf SDK
requestAsInt32String = struct.pack("!H", len(requestAsString)) # The message is concatenated with the reversed array
Для отправки сообщения SDK для Python использует метод Protocol.transport.write().

Чтение сообщения

Асинхронный код

Все SDK Open API основаны на асинхронном выполнении, что означает, что они не ожидают прибытия сообщений, а вместо этого реагируют на динамически поступающие сообщения. В результате получение сообщения обычно выполняется через обработчики событий

Для чтения сообщения вам потребуется выполнить последовательность действий, обратную шагам, необходимым для отправки сообщения.

  1. Получите первые четыре байта массива байтов (помните, они обозначают длину сообщения). Переверните эти четыре байта и преобразуйте их в целое число.

  2. Прочитайте X количество байтов из потока, где X — это целое число, полученное на шаге 1.

  3. Используйте SDK Google Protobuf для десериализации сообщения в допустимое ProtoMessage.

  4. Используйте поле payloadType объекта ProtoMessage, чтобы найти его фактический тип. С помощью SDK Google Protobuf преобразуйте ProtoMessage в объект требуемого типа ProtoOA....

Приведенные ниже фрагменты кода демонстрируют, как официальные SDK Open API подходят к чтению сообщений.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
_tcpClient = new TcpClient
{
    LingerState = new LingerOption(enable: true, 10)
};
await _tcpClient.ConnectAsync(Host, Port).ConfigureAwait(continueOnCapturedContext: false);
SslStream _sslStream = new SslStream(_tcpClient.GetStream(), leaveInnerStreamOpen: false);
await _sslStream.AuthenticateAsClientAsync(Host).ConfigureAwait(continueOnCapturedContext: false);

        private async void ReadTcp(CancellationToken cancellationToken)
    {
        byte[] dataLength = new byte[4];
        byte[] data = null;
        try
        {
            while (!IsDisposed)
            {
                int num = 0;
                do
                {
                    int count = dataLength.Length - num;
                    int num2 = num;
                    num = num2 + await _sslStream.ReadAsync(dataLength, num, count, cancellationToken).ConfigureAwait(continueOnCapturedContext: false);
                    if (num == 0)
                    {
                        throw new InvalidOperationException("Remote host closed the connection");
                    }
                }
                while (num < dataLength.Length);
                int length = GetLength(dataLength);
                if (length <= 0)
                {
                    continue;
                }

                data = ArrayPool<byte>.Shared.Rent(length);
                num = 0;
                do
                {
                    int count2 = length - num;
                    int num2 = num;
                    num = num2 + await _sslStream.ReadAsync(data, num, count2, cancellationToken).ConfigureAwait(continueOnCapturedContext: false);
                    if (num == 0)
                    {
                        throw new InvalidOperationException("Remote host closed the connection");
                    }
                }
                while (num < length);
                ProtoMessage protoMessage = ProtoMessage.Parser.ParseFrom(data, 0, length);
                ArrayPool<byte>.Shared.Return(data);
                OnNext(protoMessage);
            }
        }
        catch (Exception innerException)
        {
            if (data != null)
            {
                ArrayPool<byte>.Shared.Return(data);
            }

            ReceiveException exception = new ReceiveException(innerException);
            OnError(exception);
        }
    }

    private int GetLength(byte[] lengthBytes)
    {
        Span<byte> span = lengthBytes.AsSpan();
        span.Reverse();
        return BitConverter.ToInt32(span);
    }

Пример на Python

В примере на Python все операции с байтами при получении сообщений обрабатываются методом dataReceived(), как показано ниже.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
def dataReceived(self, recd):
    """
    Convert int prefixed strings into calls to stringReceived.
    """
    self.recvd = self.recvd + recd
    while len(self.recvd) >= self.prefixLength and not self.paused:
        length ,= struct.unpack(
            self.structFormat, self.recvd[:self.prefixLength])
        if length > self.MAX_LENGTH:
            self.lengthLimitExceeded(length)
            return
        if len(self.recvd) < length + self.prefixLength:
            break
        packet = self.recvd[self.prefixLength:length + self.prefixLength]
        self.recvd = self.recvd[length + self.prefixLength:]
        self.stringReceived(packet)

1
2
3
4
5
6
7
8
def stringReceived(self, data):
    msg = ProtoMessage()
    msg.ParseFromString(data)

    if msg.payloadType == ProtoHeartbeatEvent().payloadType:
        self.heartbeat()
    self.factory.received(msg)
    return data

Использование WebSocket

Отправка сообщения

Для отправки сообщения через соединение WebSocket выполните следующие действия:

  1. Сериализуйте сообщение в любой подходящий формат данных (например, строку).

  2. Добавьте сериализованное сообщение в очередь отправки.

Приведенные ниже примеры демонстрируют, как эти действия выполняются в официальных SDK Open API.

SDK для C# использует класс WebsocketClient, который является частью пакета Websocket.Client. Как показано ниже, метод WebsocketClient.Send() работает следующим образом.

1
2
3
4
5
public void Send(byte[] message)
{
    Websocket.Client.Validations.Validations.ValidateInput(message, "message");
    _messagesBinaryToSendQueue.Writer.TryWrite(new ArraySegment<byte>(message));
}

Как видите, клиент просто добавляет сегмент массива в очередь отправки.

SDK для Python не поддерживает стандарт WebSocket.

Получение сообщения

Чтобы получить сообщение через соединение WebSocket, выполните следующие действия:

  1. Получите данные, полученные от серверной части cTrader.

  2. Десериализуйте данные в допустимое сообщение Protobuf.

Примеры того, как это делается в официальных SDK Open API, см. во фрагментах ниже.

Для получения сообщений WebsocketClient необходимо подписать на функцию обратного вызова, которая обрабатывает действия клиента при принятии нового сообщения.

1
2
var client = new WebsocketClient();
client.MessageReceived.Subscribe(msg => {Console.WriteLine(msg);})

Перед подпиской SDK для .NET анализирует сообщение в сообщение Protobuf. Необходимые подписки добавляются в теле обратного вызова ConnectWebSocket().

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
private async Task ConnectWebSocket()
{
    DefaultInterpolatedStringHandler defaultInterpolatedStringHandler = new DefaultInterpolatedStringHandler(7, 2);
    defaultInterpolatedStringHandler.AppendLiteral("wss://");
    defaultInterpolatedStringHandler.AppendFormatted(Host);
    defaultInterpolatedStringHandler.AppendLiteral(":");
    defaultInterpolatedStringHandler.AppendFormatted(Port);
    Uri url = new Uri(defaultInterpolatedStringHandler.ToStringAndClear());
    _websocketClient = new WebsocketClient(url, () => new ClientWebSocket())
    {
        IsTextMessageConversionEnabled = false,
        ReconnectTimeout = null,
        IsReconnectionEnabled = false,
        ErrorReconnectTimeout = null
    };
    _webSocketMessageReceivedDisposable = _websocketClient.MessageReceived.Select((ResponseMessage msg) => ProtoMessage.Parser.ParseFrom(msg.Binary)).Subscribe(new Action<ProtoMessage>(OnNext));
    _webSocketDisconnectionHappenedDisposable = _websocketClient.DisconnectionHappened.Subscribe(new Action<DisconnectionInfo>(OnWebSocketDisconnectionHappened));
    await _websocketClient.StartOrFail();
}

В обратном вызове OnNext() ProtoMessage передается в обратный вызов MessageFactory.GetMessage().

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
private void OnNext(ProtoMessage protoMessage)
{
    foreach (KeyValuePair<int, IObserver<IMessage>> observer2 in _observers)
    {
        observer2.Deconstruct(out var _, out var value);
        IObserver<IMessage> observer = value;
        try
        {
            IMessage message = MessageFactory.GetMessage(protoMessage);
            if (protoMessage.HasClientMsgId || message == null)
            {
                observer.OnNext(protoMessage);
            }

            if (message != null)
            {
                observer.OnNext(message);
            }
        }
        catch (Exception innerException)
        {
            ObserverException exception = new ObserverException(innerException, observer);
            OnError(exception);
        }
    }
}

SDK для Python не поддерживает стандарт WebSocket.