Vai al contenuto

Inviare e ricevere messaggi Protobuf

Il processo di invio e ricezione di un messaggio differisce tra gli standard TCP e WebSocket. Di seguito, spieghiamo questo processo in dettaglio per entrambi i tipi di connessioni.

Usa TCP

Invia un messaggio

Per inviare un messaggio tramite una connessione TCP, fai quanto segue:

  1. Cambia il messaggio Protobuf in un array di byte (usando la codifica Protobuf) utilizzando l'SDK ufficiale di Google Protocol Buffer per il linguaggio di programmazione scelto.

  2. Ottieni la lunghezza dell'array creato durante il passo 1. Crea un nuovo array di byte da questo intero. Inverti il nuovo array di byte.

  3. Concatena il nuovo array di byte e l'array di byte contenente il messaggio Protobuf originale.

  4. Invia l'array concatenato allo stream di connessione.

Gli esempi seguenti dimostrano come questi passaggi vengono eseguiti negli SDK Open API ufficiali.

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)

Esempio Python

Usando Twisted, l'esempio Python esegue quasi le stesse operazioni di quello C#. Il client.send(request) può essere spiegato come segue.

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
Per inviare il messaggio, l'SDK Python utilizza il metodo Protocol.transport.write().

Leggi un messaggio

Codice asincrono

Tutti gli SDK Open API si basano sull'esecuzione asincrona, il che significa che non attendono l'arrivo dei messaggi ma, invece, reagiscono ai messaggi che arrivano dinamicamente. Di conseguenza, la ricezione di un messaggio viene tipicamente effettuata tramite gestori di eventi

Per leggere un messaggio, dovrai eseguire una sequenza di azioni che inverte i passaggi richiesti per l'invio di un messaggio.

  1. Ricevi i primi quattro byte di un array di byte (ricorda, denotano la lunghezza del messaggio). Inverti questi quattro byte e cambiali in un intero.

  2. Leggi X quantità di byte da uno stream dove X è l'intero che hai ottenuto durante il passo 1.

  3. Usa l'SDK Google Protobuf per deserializzare il messaggio in un ProtoMessage valido.

  4. Usa il campo payloadType dell'oggetto ProtoMessage per trovare il suo tipo effettivo. Tramite l'SDK Google Protobuf, cambia il ProtoMessage in un oggetto del tipo ProtoOA... necessario.

Gli snippet di codice seguenti dimostrano come gli SDK Open API ufficiali affrontano la lettura dei messaggi.

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

Esempio Python

Nell'esempio Python, tutte le operazioni con i byte sulla ricezione dei messaggi sono gestite dal metodo dataReceived() come mostrato di seguito.

 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

Usa WebSocket

Invia un messaggio

Per inviare un messaggio su una connessione WebSocket, esegui le seguenti azioni:

  1. Serializza il messaggio in qualsiasi formato dati adatto (ad esempio, una stringa).

  2. Aggiungi il messaggio serializzato alla tua coda di invio.

Gli esempi seguenti dimostrano come queste azioni vengono eseguite negli SDK Open API ufficiali.

L'SDK C# utilizza la classe WebsocketClient che fa parte del pacchetto Websocket.Client. Come mostrato di seguito, il metodo WebsocketClient.Send() funziona come segue.

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

Come puoi vedere, il client aggiunge semplicemente un segmento di array alla coda di invio.

L'SDK Python non supporta lo standard WebSocket.

Ricevi un messaggio

Per ricevere un messaggio su una connessione WebSocket, fai quanto segue:

  1. Recupera i dati ricevuti dal backend di cTrader.

  2. Deserializza i dati in un messaggio Protobuf valido.

Per un'illustrazione di come questo viene fatto negli SDK Open API ufficiali, vedi gli snippet seguenti.

Per ricevere messaggi, un WebsocketClient deve essere sottoscritto a una funzione di callback che gestisce ciò che il client fa all'accettazione di un nuovo messaggio.

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

Prima di sottoscriversi, l'SDK .NET analizza il messaggio in un messaggio Protobuf. Le sottoscrizioni necessarie vengono aggiunte nel corpo del 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();
}

Nel callback OnNext(), il ProtoMessage viene passato al 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);
        }
    }
}

L'SDK Python non supporta lo standard WebSocket.