Aller au contenu

Envoyer et recevoir des messages Protobuf

Le processus d'envoi et de réception d'un message diffère entre les normes TCP et WebSocket. Ci-dessous, nous expliquons ce processus en détail pour les deux types de connexions.

Utiliser TCP

Envoyer un message

Pour envoyer un message via une connexion TCP, procédez comme suit :

  1. Transformez le message Protobuf en un tableau d'octets (en utilisant l'encodage Protobuf) en utilisant le SDK officiel Google Protocol Buffer pour le langage de programmation de votre choix.

  2. Obtenez la longueur du tableau créé à l'étape 1. Créez un nouveau tableau d'octets à partir de cet entier. Inversez le nouveau tableau d'octets.

  3. Concaténez le nouveau tableau d'octets et le tableau d'octets contenant le message Protobuf original.

  4. Envoyez le tableau concaténé au flux de connexion.

Les exemples ci-dessous montrent comment ces étapes sont effectuées dans les SDK officiels de l'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)

Exemple Python

En utilisant Twisted, l'exemple Python effectue presque les mêmes opérations que celui en C#. Le client.send(request) peut être expliqué comme suit.

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
Pour envoyer le message, le SDK Python utilise la méthode Protocol.transport.write().

Lire un message

Code asynchrone

Tous les SDK Open API reposent sur une exécution asynchrone, ce qui signifie qu'ils n'attendent pas l'arrivée des messages mais, au contraire, réagissent aux messages arrivant dynamiquement. Par conséquent, la réception d'un message se fait généralement via des gestionnaires d'événements

Pour lire un message, vous devrez effectuer une séquence d'actions qui inverse les étapes requises pour envoyer un message.

  1. Recevez les quatre premiers octets d'un tableau d'octets (rappelez-vous, ils indiquent la longueur du message). Inversez ces quatre octets et transformez-les en un entier.

  2. Lisez X octets à partir d'un flux où X est l'entier que vous avez obtenu à l'étape 1.

  3. Utilisez le SDK Google Protobuf pour désérialiser le message en un ProtoMessage valide.

  4. Utilisez le champ payloadType de l'objet ProtoMessage pour trouver son type réel. Via le SDK Google Protobuf, transformez le ProtoMessage en un objet du type ProtoOA... nécessaire.

Les extraits de code ci-dessous montrent comment les SDK officiels de l'Open API abordent la lecture des messages.

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

Exemple Python

Dans l'exemple Python, toutes les opérations avec les octets lors de la réception des messages sont gérées par la méthode dataReceived() comme montré ci-dessous.

 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

Utiliser WebSocket

Envoyer un message

Pour envoyer un message via une connexion WebSocket, effectuez les actions suivantes :

  1. Sérialisez le message dans n'importe quel format de données approprié (par exemple, une chaîne).

  2. Ajoutez le message sérialisé à votre file d'attente d'envoi.

Les exemples ci-dessous montrent comment ces actions sont effectuées dans les SDK officiels de l'Open API.

Le SDK C# utilise la classe WebsocketClient qui fait partie du package Websocket.Client. Comme montré ci-dessous, la méthode WebsocketClient.Send() fonctionne comme suit.

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

Comme vous pouvez le voir, le client ajoute simplement un segment de tableau à la file d'attente d'envoi.

Le SDK Python ne prend pas en charge le standard WebSocket.

Recevoir un message

Pour recevoir un message via une connexion WebSocket, procédez comme suit :

  1. Récupérez les données reçues de l'arrière-plan de cTrader.

  2. Désérialisez les données en un message Protobuf valide.

Pour une illustration de la façon dont cela est fait dans les SDK officiels de l'Open API, voir les extraits ci-dessous.

Pour recevoir des messages, un WebsocketClient doit être abonné à une fonction de callback qui gère ce que le client fait lors de l'acceptation d'un nouveau message.

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

Avant de s'abonner, le SDK .NET analyse le message en un message Protobuf. Les abonnements nécessaires sont ajoutés dans le corps du callback 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();
}

Dans le callback OnNext(), le ProtoMessage est transmis au callback 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);
        }
    }
}

Le SDK Python ne prend pas en charge le standard WebSocket.