Przejdź do treści

Wysyłanie i odbieranie wiadomości Protobuf

Proces wysyłania i odbierania wiadomości różni się między standardami TCP i WebSocket. Poniżej szczegółowo wyjaśniamy ten proces dla obu typów połączeń.

Użyj TCP

Wyślij wiadomość

Aby wysłać wiadomość przez połączenie TCP, wykonaj następujące czynności:

  1. Zmień wiadomość Protobuf na tablicę bajtów (używając kodowania Protobuf) przy użyciu oficjalnego SDK Google Protocol Buffer dla wybranego języka programowania.

  2. Pobierz długość tablicy utworzonej podczas kroku 1. Utwórz nową tablicę bajtów z tej liczby całkowitej. Odwróć nową tablicę bajtów.

  3. Połącz nową tablicę bajtów i tablicę bajtów zawierającą oryginalną wiadomość Protobuf.

  4. Wyślij połączoną tablicę do strumienia połączenia.

Poniższe przykłady pokazują, jak te kroki są wykonywane w oficjalnych 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)

Przykład Python

Używając Twisted, przykład Python wykonuje niemal te same operacje co przykład C#. client.send(request) można wyjaśnić w następujący sposób.

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
Do wysyłania wiadomości SDK Python używa metody Protocol.transport.write().

Odczytaj wiadomość

Kod asynchroniczny

Wszystkie SDK Open API opierają się na wykonywaniu asynchronicznym, co oznacza, że nie czekają na nadejście wiadomości, ale zamiast tego reagują na dynamicznie przychodzące wiadomości. W rezultacie odbieranie wiadomości odbywa się zazwyczaj za pośrednictwem procedur obsługi zdarzeń

Aby odczytać wiadomość, będziesz musiał wykonać sekwencję działań, która odwraca kroki wymagane do wysłania wiadomości.

  1. Odbierz pierwsze cztery bajty tablicy bajtów (pamiętaj, oznaczają one długość wiadomości). Odwróć te cztery bajty i zmień je na liczbę całkowitą.

  2. Odczytaj X bajtów ze strumienia, gdzie X to liczba całkowita, którą otrzymałeś podczas kroku 1.

  3. Użyj SDK Google Protobuf, aby zdeserializować wiadomość do prawidłowej ProtoMessage.

  4. Użyj pola payloadType obiektu ProtoMessage, aby znaleźć jego rzeczywisty typ. Za pomocą Google Protobuf SDK zmień ProtoMessage na obiekt wymaganego typu ProtoOA....

Poniższe fragmenty kodu pokazują, jak oficjalne SDK Open API odczytują wiadomości.

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

Przykład w Pythonie

W przykładzie w Pythonie wszystkie operacje na bajtach podczas odbierania wiadomości są obsługiwane przez metodę dataReceived(), jak pokazano poniżej.

 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

Użyj WebSocket

Wyślij wiadomość

Aby wysłać wiadomość przez połączenie WebSocket, wykonaj następujące czynności:

  1. Serializuj wiadomość do dowolnego odpowiedniego formatu danych (na przykład ciągu znaków).

  2. Dodaj serializowaną wiadomość do kolejki wysyłania.

Poniższe przykłady pokazują, jak te czynności są wykonywane w oficjalnych SDK Open API.

SDK C# używa klasy WebsocketClient, która jest częścią pakietu Websocket.Client. Jak pokazano poniżej, metoda WebsocketClient.Send() działa w następujący sposób.

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

Jak widać, klient po prostu dodaje segment tablicy do kolejki wysyłania.

SDK Pythona nie obsługuje standardu WebSocket.

Odbierz wiadomość

Aby odebrać wiadomość przez połączenie WebSocket, wykonaj następujące czynności:

  1. Pobierz dane odebrane z backendu cTrader.

  2. Deserializuj dane do prawidłowej wiadomości Protobuf.

Aby zobaczyć ilustrację, jak to jest wykonywane w oficjalnych SDK Open API, zobacz poniższe fragmenty.

Aby odbierać wiadomości, WebsocketClient musi być subskrybowany do funkcji zwrotnej, która obsługuje to, co klient robi po przyjęciu nowej wiadomości.

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

Przed subskrypcją SDK .NET analizuje wiadomość na wiadomość Protobuf. Niezbędne subskrypcje są dodawane w treści wywołania zwrotnego 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();
}

W wywołaniu zwrotnym OnNext() ProtoMessage jest przekazywany do wywołania zwrotnego 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 Pythona nie obsługuje standardu WebSocket.