Ana içeriğe geç

Protobuf mesajları gönderin ve alın

Bir mesaj gönderme ve alma süreci, TCP ve WebSocket standartları arasında farklılık gösterir. Aşağıda, bu süreci her iki bağlantı türü için ayrıntılı olarak açıklıyoruz.

TCP Kullanın

Mesaj gönderin

Bir TCP bağlantısı aracılığıyla mesaj göndermek için aşağıdakileri yapın:

  1. Seçtiğiniz programlama dili için resmi Google Protocol Buffer SDK'sını kullanarak Protobuf mesajını bir bayt dizisine (Protobuf kodlamasını kullanarak) dönüştürün.

    1. adımda oluşturulan dizinin uzunluğunu alın. Bu tam sayıdan yeni bir bayt dizisi oluşturun. Yeni bayt dizisini ters çevirin.

  3. Yeni bayt dizisini ve orijinal Protobuf mesajını içeren bayt dizisini birleştirin.

  4. Birleştirilmiş diziyi bağlantı akışına gönderin.

Aşağıdaki örnekler, bu adımların resmi Open API SDK'larında nasıl gerçekleştirildiğini göstermektedir.

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 örneği

Twisted kullanarak, Python örneği C# örneğiyle neredeyse aynı işlemleri gerçekleştirir. client.send(request) aşağıdaki gibi açıklanabilir.

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
Mesajı göndermek için Python SDK'sı Protocol.transport.write() yöntemini kullanır.

Mesaj okuyun

Asenkron kod

Tüm Open API SDK'ları asenkron yürütmeye dayanır, yani mesajların gelmesini beklemezler, bunun yerine dinamik olarak gelen mesajlara tepki verirler. Sonuç olarak, mesaj almak genellikle olay işleyicileri aracılığıyla yapılır.

Bir mesajı okumak için, mesaj göndermek için gereken adımları tersine çeviren bir dizi eylem gerçekleştirmeniz gerekecektir.

  1. Bir bayt dizisinin ilk dört baytını alın (unutmayın, bunlar mesaj uzunluğunu belirtir). Bu dört baytı ters çevirin ve bir tam sayıya dönüştürün.

    1. adımda elde ettiğiniz tam sayı olan X kadar baytı bir akıştan okuyun.

  3. Mesajı geçerli bir ProtoMessage'a seri durumdan çıkarmak için Google Protobuf SDK'sını kullanın.

  4. ProtoMessage nesnesinin payloadType alanını kullanarak gerçek türünü bulun. Google Protobuf SDK aracılığıyla, ProtoMessage'ı gerekli ProtoOA... türünde bir nesneye dönüştürün.

Aşağıdaki kod parçacıkları, resmi Open API SDK'larının mesaj okumaya nasıl yaklaştığını göstermektedir.

 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 örneği

Python örneğinde, mesaj alırken baytlarla yapılan tüm işlemler, aşağıda gösterildiği gibi dataReceived() yöntemi tarafından ele alınır.

 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 Kullanın

Mesaj gönderin

Bir WebSocket bağlantısı üzerinden mesaj göndermek için aşağıdaki eylemleri gerçekleştirin:

  1. Mesajı uygun herhangi bir veri biçimine (örneğin, bir dize) seri hale getirin.

  2. Seri hale getirilmiş mesajı gönderme kuyruğunuza ekleyin.

Aşağıdaki örnekler, bu eylemlerin resmi Open API SDK'larında nasıl gerçekleştirildiğini göstermektedir.

C# SDK'sı, Websocket.Client paketinin bir parçası olan WebsocketClient sınıfını kullanır. Aşağıda gösterildiği gibi, WebsocketClient.Send() yöntemi aşağıdaki gibi çalışır.

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

Gördüğünüz gibi, istemci yalnızca gönderme kuyruğuna bir dizi segmenti ekler.

Python SDK'sı WebSocket standardını desteklemez.

Mesaj alın

Bir WebSocket bağlantısı üzerinden mesaj almak için aşağıdakileri yapın:

  1. cTrader arka ucundan alınan verileri alın.

  2. Verileri geçerli bir Protobuf mesajına seri durumdan çıkarın.

Bunun resmi Open API SDK'larında nasıl yapıldığına dair bir örnek için aşağıdaki kod parçacıklarına bakın.

Mesaj almak için, bir WebsocketClient'ın, istemcinin yeni bir mesajı kabul ettiğinde ne yapacağını yöneten bir geri arama işlevine abone olması gerekir.

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

Abone olmadan önce, .NET SDK'sı mesajı bir Protobuf mesajına ayrıştırır. Gerekli abonelikler ConnectWebSocket() geri aramasının gövdesine eklenir.

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

OnNext() geri aramasında, ProtoMessage MessageFactory.GetMessage() geri aramasına iletilir.

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

Python SDK'sı WebSocket standardını desteklemez.