Saltar a contenido

Enviar y recibir mensajes Protobuf

El proceso de enviar y recibir un mensaje difiere entre los estándares TCP y WebSocket. A continuación, explicamos este proceso en detalle para ambos tipos de conexiones.

Usar TCP

Enviar un mensaje

Para enviar un mensaje a través de una conexión TCP, haga lo siguiente:

  1. Cambie el mensaje Protobuf a una matriz de bytes (usando la codificación Protobuf) utilizando el SDK oficial de Google Protocol Buffer para su lenguaje de programación elegido.

  2. Obtenga la longitud de la matriz creada durante el paso 1. Cree una nueva matriz de bytes a partir de este entero. Invierta la nueva matriz de bytes.

  3. Concatene la nueva matriz de bytes y la matriz de bytes que contiene el mensaje Protobuf original.

  4. Envíe la matriz concatenada al flujo de conexión.

Los ejemplos a continuación demuestran cómo se realizan estos pasos en los SDK oficiales de 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)

Ejemplo de Python

Usando Twisted, el ejemplo de Python realiza prácticamente las mismas operaciones que el de C#. El client.send(request) se puede explicar de la siguiente manera.

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
Para enviar el mensaje, el SDK de Python utiliza el método Protocol.transport.write().

Leer un mensaje

Código asíncrono

Todos los SDK de Open API se basan en la ejecución asíncrona, lo que significa que no esperan a que lleguen los mensajes, sino que reaccionan a los mensajes que llegan dinámicamente. Como resultado, la recepción de un mensaje se realiza normalmente a través de controladores de eventos

Para leer un mensaje, deberá realizar una secuencia de acciones que invierta los pasos necesarios para enviar un mensaje.

  1. Reciba los primeros cuatro bytes de una matriz de bytes (recuerde, indican la longitud del mensaje). Invierta estos cuatro bytes y conviértalos en un número entero.

  2. Lea X cantidad de bytes de un flujo donde X es el número entero que obtuvo durante el paso 1.

  3. Utilice el SDK de Google Protobuf para deserializar el mensaje en un ProtoMessage válido.

  4. Utilice el campo payloadType del objeto ProtoMessage para encontrar su tipo real. A través del SDK de Google Protobuf, cambie el ProtoMessage a un objeto del tipo ProtoOA... necesario.

Los fragmentos de código a continuación demuestran cómo los SDK oficiales de Open API abordan la lectura de mensajes.

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

Ejemplo de Python

En el ejemplo de Python, todas las operaciones con bytes al recibir mensajes son manejadas por el método dataReceived() como se muestra a continuación.

 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

Usar WebSocket

Enviar un mensaje

Para enviar un mensaje a través de una conexión WebSocket, realice las siguientes acciones:

  1. Serialice el mensaje en cualquier formato de datos adecuado (por ejemplo, una cadena).

  2. Agregue el mensaje serializado a su cola de envío.

Los ejemplos a continuación demuestran cómo se realizan estas acciones en los SDK oficiales de Open API.

El SDK de C# utiliza la clase WebsocketClient que es parte del paquete Websocket.Client. Como se muestra a continuación, el método WebsocketClient.Send() funciona de la siguiente manera.

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

Como puede ver, el cliente simplemente agrega un segmento de matriz a la cola de envío.

El SDK de Python no admite el estándar WebSocket.

Recibir un mensaje

Para recibir un mensaje a través de una conexión WebSocket, haga lo siguiente:

  1. Recupere los datos recibidos del backend de cTrader.

  2. Deserialice los datos en un mensaje Protobuf válido.

Para ver una ilustración de cómo se hace esto en los SDK oficiales de Open API, vea los fragmentos a continuación.

Para recibir mensajes, un WebsocketClient necesita estar suscrito a una función de devolución de llamada que maneje lo que hace el cliente al aceptar un nuevo mensaje.

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

Antes de suscribirse, el SDK de .NET analiza el mensaje en un mensaje Protobuf. Las suscripciones necesarias se agregan en el cuerpo de la devolución de llamada 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();
}

En la devolución de llamada OnNext(), el ProtoMessage se pasa a la devolución de llamada 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);
        }
    }
}

El SDK de Python no admite el estándar WebSocket.