Zum Inhalt

Protobuf-Nachrichten senden und empfangen

Der Prozess des Sendens und Empfangens einer Nachricht unterscheidet sich zwischen den TCP- und WebSocket-Standards. Im Folgenden erklären wir diesen Prozess ausführlich für beide Verbindungstypen.

TCP verwenden

Eine Nachricht senden

Um eine Nachricht über eine TCP-Verbindung zu senden, gehen Sie wie folgt vor:

  1. Ändern Sie die Protobuf-Nachricht in ein Byte-Array (mithilfe der Protobuf-Codierung), indem Sie das offizielle Google Protocol Buffer SDK für Ihre gewählte Programmiersprache verwenden.

  2. Ermitteln Sie die Länge des in Schritt 1 erstellten Arrays. Erstellen Sie aus dieser Ganzzahl ein neues Byte-Array. Kehren Sie das neue Byte-Array um.

  3. Verketten Sie das neue Byte-Array und das Byte-Array, das die ursprüngliche Protobuf-Nachricht enthält.

  4. Senden Sie das verkettete Array an den Verbindungsstream.

Die folgenden Beispiele zeigen, wie diese Schritte in den offiziellen Open API SDKs ausgeführt werden.

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-Beispiel

Mithilfe von Twisted führt das Python-Beispiel nahezu dieselben Vorgänge wie das C#-Beispiel aus. Das client.send(request) kann wie folgt erklärt werden.

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
Zum Senden der Nachricht verwendet das Python SDK die Methode Protocol.transport.write().

Eine Nachricht lesen

Asynchroner Code

Alle Open API SDKs basieren auf asynchroner Ausführung, d. h., sie warten nicht auf eingehende Nachrichten, sondern reagieren stattdessen auf dynamisch eintreffende Nachrichten. Daher erfolgt der Empfang einer Nachricht in der Regel über Ereignishandler

Um eine Nachricht zu lesen, müssen Sie eine Abfolge von Aktionen ausführen, die die zum Senden einer Nachricht erforderlichen Schritte umkehrt.

  1. Empfangen Sie die ersten vier Bytes eines Byte-Arrays (denken Sie daran, dass sie die Nachrichtenlänge angeben). Kehren Sie diese vier Bytes um und ändern Sie sie in eine Ganzzahl.

  2. Lesen Sie X Bytes aus einem Stream, wobei X die Ganzzahl ist, die Sie in Schritt 1 erhalten haben.

  3. Verwenden Sie das Google Protobuf SDK, um die Nachricht in eine gültige ProtoMessage zu deserialisieren.

  4. Verwenden Sie das Feld payloadType des Objekts ProtoMessage, um dessen tatsächlichen Typ zu ermitteln. Ändern Sie über das Google Protobuf SDK die ProtoMessage in ein Objekt des benötigten Typs ProtoOA....

Die folgenden Codeausschnitte zeigen, wie die offiziellen Open API SDKs das Lesen von Nachrichten handhaben.

 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-Beispiel

Im Python-Beispiel werden alle Vorgänge mit Bytes beim Empfangen von Nachrichten von der Methode dataReceived() verarbeitet, wie unten gezeigt.

 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 verwenden

Eine Nachricht senden

Um eine Nachricht über eine WebSocket-Verbindung zu senden, führen Sie die folgenden Aktionen aus:

  1. Serialisieren Sie die Nachricht in ein beliebiges geeignetes Datenformat (z. B. eine Zeichenfolge).

  2. Fügen Sie die serialisierte Nachricht Ihrer Sendewarteschlange hinzu.

Die folgenden Beispiele zeigen, wie diese Aktionen in den offiziellen Open API SDKs ausgeführt werden.

Das C# SDK verwendet die Klasse WebsocketClient, die Teil des Pakets Websocket.Client ist. Wie unten gezeigt, funktioniert die Methode WebsocketClient.Send() wie folgt.

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

Wie Sie sehen können, fügt der Client der Sendewarteschlange einfach ein Array-Segment hinzu.

Das Python SDK unterstützt den WebSocket-Standard nicht.

Eine Nachricht empfangen

Um eine Nachricht über eine WebSocket-Verbindung zu empfangen, gehen Sie wie folgt vor:

  1. Rufen Sie die vom cTrader-Backend empfangenen Daten ab.

  2. Deserialisieren Sie die Daten in eine gültige Protobuf-Nachricht.

Eine Veranschaulichung, wie dies in den offiziellen Open API SDKs erfolgt, finden Sie in den folgenden Ausschnitten.

Um Nachrichten zu empfangen, muss ein WebsocketClient eine Rückruffunktion abonnieren, die festlegt, was der Client beim Akzeptieren einer neuen Nachricht tut.

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

Vor dem Abonnieren analysiert das .NET SDK die Nachricht in eine Protobuf-Nachricht. Die erforderlichen Abonnements werden im Hauptteil des Rückrufs ConnectWebSocket() hinzugefügt.

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

Im Rückruf OnNext() wird die ProtoMessage an den Rückruf MessageFactory.GetMessage() übergeben.

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

Das Python SDK unterstützt den WebSocket-Standard nicht.