Multiplex TCP Streaming Example

This example demonstrates the complete Subject Interface v1 workflow, showcasing both the JSON-REST API layer for session management and the TCP Streaming Protocol layer for real-time data exchange. The implementation illustrates how clients interact with the dual-layer architecture to establish streaming sessions and exchange C-ITS messages.

TLC Terminology

The interface originally served Traffic Light Controllers (TLCs), and this legacy terminology persists in the API specifications for backward compatibility. "TLC" now refers to any connected entity: traffic controllers, sensors, barriers, displays, or any C-ITS participant. See Terminology Reference for the mapping between legacy and generic terminology.

The example runs two concurrent sessions using the multiplex protocol:

A TLC session representing a connected subject that produces data
A Broker session acting as a data distribution hub that consumes and routes messages

While this example shows a simplified scenario with the TLC session sending data and the Broker session receiving it, the Subject Interface supports full bi-directional communication. TLC sessions can receive data from Broker sessions, and Broker sessions can send data to TLC sessions.

The implementation demonstrates the essential workflow: authenticating with the REST API, creating streaming sessions, establishing TCP connections to the Streaming Service Nodes, and exchanging framed messages using the protocol's datagram format.

Important Disclaimer

This example is for educational purposes only and is not production-ready. It demonstrates the basic workflow and protocol interaction but lacks the error handling, reconnection logic, monitoring, and robustness required for production deployments. Use this as a learning resource and starting point for understanding the Subject Interface v1.

Prerequisites

General Requirements

Valid authorization tokens for Subject Interface v1 API access
Access to a Subject Interface v1 instance
Network connectivity to both REST API and TCP streaming endpoints

Language-Specific Requirements

PythonGoJavaRust.NET

Python 3.10 or later (uses match/case statements)
requests library for HTTP API calls
Standard library modules (socket, ssl, struct, threading, json, logging)

Go 1.21 or later
Standard library modules (net/http, crypto/tls, encoding/binary, sync)

Java 11 or later
org.json library (automatically downloaded in Docker build)
Standard library modules (java.net, javax.net.ssl, java.util.concurrent)

Rust 1.83 or later
Cargo dependencies: reqwest, serde, tokio, rustls, rand

.NET 8.0 or later
NuGet dependencies: Newtonsoft.Json

Configuration by Environment

The example uses environment variables for configuration:

Variable	Description	Example
`STREAMING_API_BASEURL`	Base URL of the Subject Interface API	`https://localhost/api`
`STREAMING_API_TLC_TOKEN`	Authorization token for TLC session	`your-tlc-auth-token`
`STREAMING_API_BROKER_TOKEN`	Authorization token for Broker session	`your-broker-auth-token`
`STREAMING_API_DOMAIN`	Domain for session creation	`dev_001`
`STREAMING_API_SECURITY_MODE`	Security mode (`NONE` or `TLSv1.2`)	`TLSv1.2`
`STREAMING_API_IDENTIFIER`	TLC identifier for payload messages	`sub00001`

PythonGoJavaRust.NET

STREAMING_API_BASEURL = os.environ.get("STREAMING_API_BASEURL", "https://localhost/api")
STREAMING_API_TLC_TOKEN = os.environ.get("STREAMING_API_TLC_TOKEN", "your-tlc-auth-token")
STREAMING_API_BROKER_TOKEN = os.environ.get("STREAMING_API_BROKER_TOKEN", "your-broker-auth-token")
STREAMING_API_DOMAIN = os.environ.get("STREAMING_API_DOMAIN", "dev_001")
STREAMING_API_SECURITY_MODE = os.environ.get("STREAMING_API_SECURITY_MODE", "TLSv1.2")
STREAMING_API_IDENTIFIER = os.environ.get("STREAMING_API_IDENTIFIER", "sub00001")

var (
    STREAMING_API_BASEURL      = getEnv("STREAMING_API_BASEURL", "https://localhost/api")
    STREAMING_API_TLC_TOKEN    = getEnv("STREAMING_API_TLC_TOKEN", "your-tlc-auth-token")
    STREAMING_API_BROKER_TOKEN = getEnv("STREAMING_API_BROKER_TOKEN", "your-broker-auth-token")
    STREAMING_API_DOMAIN       = getEnv("STREAMING_API_DOMAIN", "dev_001")
    STREAMING_API_SECURITY_MODE = getEnv("STREAMING_API_SECURITY_MODE", "TLSv1.2")
    STREAMING_API_IDENTIFIER   = getEnv("STREAMING_API_IDENTIFIER", "sub00001")
)

    private static final String STREAMING_API_BASEURL = System.getenv("STREAMING_API_BASEURL") != null 
        ? System.getenv("STREAMING_API_BASEURL") : "https://localhost/api";
    private static final String STREAMING_API_TLC_TOKEN = System.getenv("STREAMING_API_TLC_TOKEN") != null 
        ? System.getenv("STREAMING_API_TLC_TOKEN") : "your-tlc-auth-token";
    private static final String STREAMING_API_BROKER_TOKEN = System.getenv("STREAMING_API_BROKER_TOKEN") != null 
        ? System.getenv("STREAMING_API_BROKER_TOKEN") : "your-broker-auth-token";
    private static final String STREAMING_API_DOMAIN = System.getenv("STREAMING_API_DOMAIN") != null 
        ? System.getenv("STREAMING_API_DOMAIN") : "dev_001";
    private static final String STREAMING_API_SECURITY_MODE = System.getenv("STREAMING_API_SECURITY_MODE") != null 
        ? System.getenv("STREAMING_API_SECURITY_MODE") : "TLSv1.2";
    private static final String STREAMING_API_IDENTIFIER = System.getenv("STREAMING_API_IDENTIFIER") != null 
        ? System.getenv("STREAMING_API_IDENTIFIER") : "sub00001";

fn get_env(key: &str, default: &str) -> String {
    env::var(key).unwrap_or_else(|_| default.to_string())
}

lazy_static::lazy_static! {
    static ref STREAMING_API_BASEURL: String = get_env("STREAMING_API_BASEURL", "https://localhost/api");
    static ref STREAMING_API_TLC_TOKEN: String = get_env("STREAMING_API_TLC_TOKEN", "your-tlc-auth-token");
    static ref STREAMING_API_BROKER_TOKEN: String = get_env("STREAMING_API_BROKER_TOKEN", "your-broker-auth-token");
    static ref STREAMING_API_DOMAIN: String = get_env("STREAMING_API_DOMAIN", "dev_001");
    static ref STREAMING_API_SECURITY_MODE: String = get_env("STREAMING_API_SECURITY_MODE", "TLSv1.2");
    static ref STREAMING_API_IDENTIFIER: String = get_env("STREAMING_API_IDENTIFIER", "sub00001");
}

public static class Config
{
    public static string StreamingApiBaseUrl { get; } = Environment.GetEnvironmentVariable("STREAMING_API_BASEURL") ?? "https://localhost/api";
    public static string StreamingApiTlcToken { get; } = Environment.GetEnvironmentVariable("STREAMING_API_TLC_TOKEN") ?? "your-tlc-auth-token";
    public static string StreamingApiBrokerToken { get; } = Environment.GetEnvironmentVariable("STREAMING_API_BROKER_TOKEN") ?? "your-broker-auth-token";
    public static string StreamingApiDomain { get; } = Environment.GetEnvironmentVariable("STREAMING_API_DOMAIN") ?? "dev_001";
    public static string StreamingApiSecurityMode { get; } = Environment.GetEnvironmentVariable("STREAMING_API_SECURITY_MODE") ?? "TLSv1.2";
    public static string StreamingApiIdentifier { get; } = Environment.GetEnvironmentVariable("STREAMING_API_IDENTIFIER") ?? "sub00001";
}

REST API Client Functions

The REST API client creates sessions for both TLC and Broker types:

PythonGoJavaRust.NET

def create_session(type, token, api_url, security_mode, identifier): 
    url = f"{api_url}/v1/sessions"
    headers = {
        'X-Authorization': f"{token}",
        'Content-Type': 'application/json'
    }

    data = {
        "domain": f"{STREAMING_API_DOMAIN}",
        "type": type,
        "protocol": "TCPStreaming_Multiplex",
        "details": {
            "securityMode": security_mode,
            "tlcIdentifiers": [identifier]
        }
    }

    response = requests.post(url, headers=headers, json=data)

    if response.ok:
        response_data = response.json()
        log_json("Session created successfully", response_data)
        token = response_data.get('token')
        host = response_data['details']['listener']['host']
        port = response_data['details']['listener']['port']
        log(f"Token: {token}")
        log(f"Host: {host}")
        log(f"Port: {port}")
        return host, port, token
    else:
        log(f"Request failed with status code {response.status_code}")

func createSession(sessionType, token, apiURL, securityMode, identifier, threadName string) (string, int, string, error) {
    url := fmt.Sprintf("%s/v1/sessions", apiURL)

    requestData := map[string]interface{}{
        "domain":   STREAMING_API_DOMAIN,
        "type":     sessionType,
        "protocol": "TCPStreaming_Multiplex",
        "details": map[string]interface{}{
            "securityMode":   securityMode,
            "tlcIdentifiers": []string{identifier},
        },
    }

    jsonData, err := json.Marshal(requestData)
    if err != nil {
        return "", 0, "", fmt.Errorf("failed to marshal request data: %v", err)
    }

    req, err := http.NewRequest("POST", url, bytes.NewBuffer(jsonData))
    if err != nil {
        return "", 0, "", fmt.Errorf("failed to create request: %v", err)
    }

    req.Header.Set("X-Authorization", token)
    req.Header.Set("Content-Type", "application/json")

    client := &http.Client{}
    resp, err := client.Do(req)
    if err != nil {
        return "", 0, "", fmt.Errorf("request failed: %v", err)
    }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        return "", 0, "", fmt.Errorf("request failed with status code %d", resp.StatusCode)
    }

    var responseData map[string]interface{}
    if err := json.NewDecoder(resp.Body).Decode(&responseData); err != nil {
        return "", 0, "", fmt.Errorf("failed to decode response: %v", err)
    }

    logJSON(threadName, "Session created successfully", responseData)

    sessionToken := responseData["token"].(string)
    host := responseData["details"].(map[string]interface{})["listener"].(map[string]interface{})["host"].(string)
    port := int(responseData["details"].(map[string]interface{})["listener"].(map[string]interface{})["port"].(float64))

    logMsg(threadName, fmt.Sprintf("Token: %s", sessionToken))
    logMsg(threadName, fmt.Sprintf("Host: %s", host))
    logMsg(threadName, fmt.Sprintf("Port: %d", port))

    return host, port, sessionToken, nil
}

private static class SessionInfo {
    String host;
    int port;
    String token;

    SessionInfo(String host, int port, String token) {
        this.host = host;
        this.port = port;
        this.token = token;
    }
}

private static SessionInfo createSession(String type, String token, String apiUrl, 
        String securityMode, String identifier, String threadName) throws Exception {
    String url = apiUrl + "/v1/sessions";

    JSONObject requestData = new JSONObject();
    requestData.put("domain", STREAMING_API_DOMAIN);
    requestData.put("type", type);
    requestData.put("protocol", "TCPStreaming_Multiplex");

    JSONObject details = new JSONObject();
    details.put("securityMode", securityMode);
    details.put("tlcIdentifiers", new JSONArray().put(identifier));
    requestData.put("details", details);

    // Disable SSL verification for localhost (development only)
    if (apiUrl.contains("localhost")) {
        disableSSLVerification();
    }

    HttpURLConnection conn = (HttpURLConnection) new URL(url).openConnection();
    conn.setRequestMethod("POST");
    conn.setRequestProperty("X-Authorization", token);
    conn.setRequestProperty("Content-Type", "application/json");
    conn.setDoOutput(true);

    try (OutputStreamWriter writer = new OutputStreamWriter(conn.getOutputStream())) {
        writer.write(requestData.toString());
    }

    int responseCode = conn.getResponseCode();
    if (responseCode != HttpURLConnection.HTTP_OK) {
        throw new Exception("Request failed with status code " + responseCode);
    }

    StringBuilder response = new StringBuilder();
    try (BufferedReader reader = new BufferedReader(new InputStreamReader(conn.getInputStream()))) {
        String line;
        while ((line = reader.readLine()) != null) {
            response.append(line);
        }
    }

    JSONObject responseData = new JSONObject(response.toString());
    logJSON(threadName, "Session created successfully", responseData);

    String sessionToken = responseData.getString("token");
    String host = responseData.getJSONObject("details").getJSONObject("listener").getString("host");
    int port = responseData.getJSONObject("details").getJSONObject("listener").getInt("port");

    log(threadName, "Token: " + sessionToken);
    log(threadName, "Host: " + host);
    log(threadName, "Port: " + port);

    return new SessionInfo(host, port, sessionToken);
}

private static void disableSSLVerification() throws Exception {
    TrustManager[] trustAllCerts = new TrustManager[] {
        new X509TrustManager() {
            public java.security.cert.X509Certificate[] getAcceptedIssuers() { return null; }
            public void checkClientTrusted(java.security.cert.X509Certificate[] certs, String authType) {}
            public void checkServerTrusted(java.security.cert.X509Certificate[] certs, String authType) {}
        }
    };

    SSLContext sc = SSLContext.getInstance("SSL");
    sc.init(null, trustAllCerts, new java.security.SecureRandom());
    HttpsURLConnection.setDefaultSSLSocketFactory(sc.getSocketFactory());
    HttpsURLConnection.setDefaultHostnameVerifier((hostname, session) -> true);
}

#[derive(Serialize)]
struct SessionRequest {
    domain: String,
    #[serde(rename = "type")]
    session_type: String,
    protocol: String,
    details: SessionDetails,
}

#[derive(Serialize)]
struct SessionDetails {
    #[serde(rename = "securityMode")]
    security_mode: String,
    #[serde(rename = "tlcIdentifiers")]
    tlc_identifiers: Vec<String>,
}

#[derive(Deserialize)]
struct SessionResponse {
    token: String,
    details: SessionResponseDetails,
}

#[derive(Deserialize)]
struct SessionResponseDetails {
    listener: ListenerDetails,
}

#[derive(Deserialize)]
struct ListenerDetails {
    host: String,
    port: u16,
}

async fn create_session(
    session_type: &str,
    token: &str,
    api_url: &str,
    security_mode: &str,
    identifier: &str,
    thread_name: &str,
) -> Result<(String, u16, String), Box<dyn std::error::Error + Send + Sync>> {
    let url = format!("{}/v1/sessions", api_url);

    let request_data = SessionRequest {
        domain: STREAMING_API_DOMAIN.clone(),
        session_type: session_type.to_string(),
        protocol: "TCPStreaming_Multiplex".to_string(),
        details: SessionDetails {
            security_mode: security_mode.to_string(),
            tlc_identifiers: vec![identifier.to_string()],
        },
    };

    let client = reqwest::Client::new();
    let response = client
        .post(&url)
        .header("X-Authorization", token)
        .header("Content-Type", "application/json")
        .json(&request_data)
        .send()
        .await?;

    if !response.status().is_success() {
        return Err(format!("Request failed with status code {}", response.status()).into());
    }

    let response_data: Value = response.json().await?;
    log_json(thread_name, "Session created successfully", &response_data);

    let session_response: SessionResponse = serde_json::from_value(response_data)?;

    log_msg(thread_name, &format!("Token: {}", session_response.token));
    log_msg(thread_name, &format!("Host: {}", session_response.details.listener.host));
    log_msg(thread_name, &format!("Port: {}", session_response.details.listener.port));

    Ok((
        session_response.details.listener.host,
        session_response.details.listener.port,
        session_response.token,
    ))
}

public class SessionRequest
{
    [JsonProperty("domain")]
    public string Domain { get; set; } = "";

    [JsonProperty("type")]
    public string Type { get; set; } = "";

    [JsonProperty("protocol")]
    public string Protocol { get; set; } = "";

    [JsonProperty("details")]
    public SessionDetails Details { get; set; } = new();
}

public class SessionDetails
{
    [JsonProperty("securityMode")]
    public string SecurityMode { get; set; } = "";

    [JsonProperty("tlcIdentifiers")]
    public List<string> TlcIdentifiers { get; set; } = new();
}

public class SessionResponse
{
    [JsonProperty("token")]
    public string Token { get; set; } = "";

    [JsonProperty("details")]
    public SessionResponseDetails Details { get; set; } = new();
}

public class SessionResponseDetails
{
    [JsonProperty("listener")]
    public ListenerDetails Listener { get; set; } = new();
}

public class ListenerDetails
{
    [JsonProperty("host")]
    public string Host { get; set; } = "";

    [JsonProperty("port")]
    public int Port { get; set; }
}

public static class RestApi
{
    private static readonly HttpClient httpClient = new();

    public static async Task<(string host, int port, string token)> CreateSessionAsync(
        string sessionType, 
        string token, 
        string apiUrl, 
        string securityMode, 
        string identifier, 
        string threadName)
    {
        var url = $"{apiUrl}/v1/sessions";

        var requestData = new SessionRequest
        {
            Domain = Config.StreamingApiDomain,
            Type = sessionType,
            Protocol = "TCPStreaming_Multiplex",
            Details = new SessionDetails
            {
                SecurityMode = securityMode,
                TlcIdentifiers = new List<string> { identifier }
            }
        };

        var json = JsonConvert.SerializeObject(requestData);
        var content = new StringContent(json, Encoding.UTF8, "application/json");

        var request = new HttpRequestMessage(HttpMethod.Post, url)
        {
            Content = content
        };
        request.Headers.Add("X-Authorization", token);

        var response = await httpClient.SendAsync(request);

        if (!response.IsSuccessStatusCode)
        {
            throw new HttpRequestException($"Request failed with status code {response.StatusCode}");
        }

        var responseJson = await response.Content.ReadAsStringAsync();
        var responseData = JsonConvert.DeserializeObject<object>(responseJson);
        Utilities.LogJson(threadName, "Session created successfully", responseData!);

        var sessionResponse = JsonConvert.DeserializeObject<SessionResponse>(responseJson);
        if (sessionResponse == null)
        {
            throw new InvalidOperationException("Failed to deserialize session response");
        }

        Utilities.LogMessage(threadName, $"Token: {sessionResponse.Token}");
        Utilities.LogMessage(threadName, $"Host: {sessionResponse.Details.Listener.Host}");
        Utilities.LogMessage(threadName, $"Port: {sessionResponse.Details.Listener.Port}");

        return (sessionResponse.Details.Listener.Host, sessionResponse.Details.Listener.Port, sessionResponse.Token);
    }
}

Explanation

PythonGoJavaRust.NET

The create_session function creates either a TLC or Broker session with multiplex protocol support. It returns the TCP streaming endpoint details (host, port) and session token for authentication.

The createSession function creates either a TLC or Broker session with multiplex protocol support. It returns the TCP streaming endpoint details (host, port) and session token for authentication.

The create_session function creates either a TLC or Broker session with multiplex protocol support. It returns the TCP streaming endpoint details (host, port) and session token for authentication.

The CreateSessionAsync function creates either a TLC or Broker session with multiplex protocol support. It returns the TCP streaming endpoint details (host, port) and session token for authentication.

TCP Streaming Client Functions

The TCP streaming client handles the multiplex protocol communication:

PythonGoJavaRust.NET

def connect(host, port, tls):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((host, port))

    log(f"Connected to {host}:{port} (TLS: {tls})")

    if tls:
        # Wrap the socket in an SSL context and perform the TLS handshake
        sslsock = SSL_CONTEXT.wrap_socket(sock, server_hostname=host)
        sslsock.do_handshake()

        peer_name = sslsock.getpeername()
        peer_cert = sslsock.getpeercert()
        (cipher_name, cipher_version, cipher_bits) = sslsock.cipher()

        log(f"TLS handshake successful - Peer: {peer_name}, Certificate: {peer_cert}, Cipher: {cipher_name} {cipher_version} {cipher_bits} bits")

        return sslsock
    else:
        return sock


def handshake(sock):
    protocol_version_byte = b'\x01'
    sock.setblocking(True)
    sock.sendall(protocol_version_byte)
    recv = sock.recv(1)
    log(f"Received protocol version {as_hexstream(recv)}")
    if recv[0] != 1:
        raise Exception("Unsupported protocol version received")


def write_datagram(sock, datagram):
    datagram_size = len(datagram).to_bytes(2, 'big')
    header = HEADER_PREFIX + datagram_size
    frame = header + datagram
    log(f"Writing frame {as_hexstream(frame)}")
    sock.setblocking(True)
    sock.sendall(frame)


def write_token(sock, token):
    log(f"Writing token {token}")
    datagram_type = b'\x01'
    data = token.encode('ascii')
    datagram = datagram_type + data
    write_datagram(sock, datagram)


def write_keepalive(sock):
    log("Writing keep alive")
    datagram_type = b'\x00'
    data = b''
    datagram = datagram_type + data
    write_datagram(sock, datagram)


def write_timestamp_response(sock, timestamp_t0, timestamp_t1):
    timestamp_t2 = current_timestamp()
    log(f"Writing timestamp response (t0: {timestamp_t0}, t1: {timestamp_t1}, t2: {timestamp_t2})")
    datagram = struct.pack(">cQQQ", b'\x07', timestamp_t0, timestamp_t1, timestamp_t2)
    write_datagram(sock, datagram)


def write_payload_with_identifier(sock, identifier, payload_type, payload):
    log(f"Writing payload with identifier (identifier: {identifier}, payload_type: {as_hexstream(struct.pack('B', payload_type))}): {as_hexstream(payload)}")
    datagram = b'\x05' + bytes(identifier, 'ascii') + struct.pack(">BQ", payload_type, current_timestamp()) + payload
    write_datagram(sock, datagram)


def read_datagram(sock):
    sock.setblocking(False)
    try:
        header = sock.recv(4)
    except ssl.SSLWantReadError:
        # SSLSocket is not ready for reading, return None.
        return None
    except socket.error as e:
        if e.errno == errno.EAGAIN or e.errno == errno.EWOULDBLOCK:
            # No error, but socket is not ready for reading.
            return None
        else:
            raise e
    else:
        if len(header) == 0:
            raise Exception("Socked disconnected")
        log(f"Received header {as_hexstream(header)}")
        if len(header) != 4:
            raise Exception("Received partial header (implementation does not support partial receives yet)")
        prefix = header[0:2]
        if prefix != HEADER_PREFIX:
            raise Exception(f"Framing error: header prefix {as_hexstream(prefix)} != {as_hexstream(HEADER_PREFIX)}")
        size = struct.unpack(">H", header[2:4])[0]
        log(f"Trying to read {size} bytes datagram")
        sock.setblocking(True)
        datagram = sock.recv(size)
        log(f"Received datagram {as_hexstream(datagram)}")
        if len(datagram) != size:
            raise Exception("Received partial datagram (implementation does not support partial receives yet)")
        return datagram


def handle_keepalive(sock):
    log("Keep alive received")


def handle_bye(sock, datagram):
    log("Bye received")
    reason = datagram[1:]
    log(f"Bye reason: {reason}")


def handle_payload_with_identifier(sock, datagram, payload_received_callback):
    log("Payload with identifier received")
    identifier = datagram[1:9].decode('ascii')
    payload_type = datagram[9:10][0]
    origin_timestamp = struct.unpack(">Q", datagram[10:18])[0]
    payload = datagram[18:]
    log(f"Payload received (identifier: {identifier}, payload_type: {as_hexstream(struct.pack('B', payload_type))}, origin_timestamp: {origin_timestamp}): {as_hexstream(payload)}")
    payload_received_callback(identifier, payload_type, origin_timestamp, payload)


def handle_timestamp_request(sock, datagram):
    log("Timestamp request received")
    timestamp_t0 = struct.unpack(">Q", datagram[1:])[0]
    timestamp_t1 = current_timestamp()
    log(f"Timestamp request delta: {timestamp_t1 - timestamp_t0}ms")
    write_timestamp_response(sock, timestamp_t0, timestamp_t1)


def handle_datagram(sock, datagram, payload_received_callback):
    datagram_type = datagram[0:1]
    match datagram_type:
        case b'\x00':
            handle_keepalive(sock)
        case b'\x02':
            handle_bye(sock, datagram)
        case b'\x05':
            handle_payload_with_identifier(sock, datagram, payload_received_callback)
        case b'\x06':
            handle_timestamp_request(sock, datagram)
        case _:
            log(f"Unknown/unimplemented datagram type {as_hexstream(datagram_type)} received")


def run_streaming_client(host, port, session_token, tls, payload_received_callback, loop_callback):
    sock = connect(host, port, tls)

    try:
        handshake(sock)
        write_token(sock, session_token)

        while True:
            datagram = read_datagram(sock)
            if datagram:
                handle_datagram(sock, datagram, payload_received_callback)
            else:
                time.sleep(0.01)

            loop_callback(sock)

    finally:
        sock.close()

func connect(host string, port int, useTLS bool, threadName string) (net.Conn, error) {
    address := fmt.Sprintf("%s:%d", host, port)

    if useTLS {
        config := &tls.Config{
            ServerName: host,
        }
        conn, err := tls.Dial("tcp", address, config)
        if err != nil {
            return nil, fmt.Errorf("TLS connection failed: %v", err)
        }

        logMsg(threadName, fmt.Sprintf("Connected to %s:%d (TLS: %t)", host, port, useTLS))

        // Log TLS connection details
        state := conn.ConnectionState()
        logMsg(threadName, fmt.Sprintf("TLS handshake successful - Version: %x, Cipher: %x", state.Version, state.CipherSuite))

        return conn, nil
    } else {
        conn, err := net.Dial("tcp", address)
        if err != nil {
            return nil, fmt.Errorf("TCP connection failed: %v", err)
        }

        logMsg(threadName, fmt.Sprintf("Connected to %s:%d (TLS: %t)", host, port, useTLS))
        return conn, nil
    }
}

func handshake(conn net.Conn, threadName string) error {
    protocolVersionByte := []byte{0x01}
    _, err := conn.Write(protocolVersionByte)
    if err != nil {
        return fmt.Errorf("failed to send protocol version: %v", err)
    }

    recv := make([]byte, 1)
    _, err = io.ReadFull(conn, recv)
    if err != nil {
        return fmt.Errorf("failed to receive protocol version: %v", err)
    }

    logMsg(threadName, fmt.Sprintf("Received protocol version %s", asHexStream(recv)))

    if recv[0] != 1 {
        return fmt.Errorf("unsupported protocol version received")
    }

    return nil
}

func writeDatagram(conn net.Conn, datagram []byte, threadName string) error {
    datagramSize := make([]byte, 2)
    binary.BigEndian.PutUint16(datagramSize, uint16(len(datagram)))

    header := append(HEADER_PREFIX, datagramSize...)
    frame := append(header, datagram...)

    logMsg(threadName, fmt.Sprintf("Writing frame %s", asHexStream(frame)))

    _, err := conn.Write(frame)
    return err
}

func writeToken(conn net.Conn, token, threadName string) error {
    logMsg(threadName, fmt.Sprintf("Writing token %s", token))
    datagramType := []byte{0x01}
    data := []byte(token)
    datagram := append(datagramType, data...)
    return writeDatagram(conn, datagram, threadName)
}

func writeKeepalive(conn net.Conn, threadName string) error {
    logMsg(threadName, "Writing keep alive")
    datagramType := []byte{0x00}
    data := []byte{}
    datagram := append(datagramType, data...)
    return writeDatagram(conn, datagram, threadName)
}

func writeTimestampResponse(conn net.Conn, timestampT0, timestampT1 int64, threadName string) error {
    timestampT2 := currentTimestamp()
    logMsg(threadName, fmt.Sprintf("Writing timestamp response (t0: %d, t1: %d, t2: %d)", timestampT0, timestampT1, timestampT2))

    datagram := make([]byte, 25)
    datagram[0] = 0x07
    binary.BigEndian.PutUint64(datagram[1:9], uint64(timestampT0))
    binary.BigEndian.PutUint64(datagram[9:17], uint64(timestampT1))
    binary.BigEndian.PutUint64(datagram[17:25], uint64(timestampT2))

    return writeDatagram(conn, datagram, threadName)
}

func writePayloadWithIdentifier(conn net.Conn, identifier string, payloadType byte, payload []byte, threadName string) error {
    logMsg(threadName, fmt.Sprintf("Writing payload with identifier (identifier: %s, payload_type: %s): %s", 
        identifier, asHexStream([]byte{payloadType}), asHexStream(payload)))

    datagram := []byte{0x05}
    datagram = append(datagram, []byte(identifier)...)

    // Add payload type and timestamp
    payloadTypeAndTimestamp := make([]byte, 9)
    payloadTypeAndTimestamp[0] = payloadType
    binary.BigEndian.PutUint64(payloadTypeAndTimestamp[1:9], uint64(currentTimestamp()))
    datagram = append(datagram, payloadTypeAndTimestamp...)
    datagram = append(datagram, payload...)

    return writeDatagram(conn, datagram, threadName)
}

func readDatagram(conn net.Conn, threadName string) ([]byte, error) {
    // Set read timeout for non-blocking behavior
    conn.SetReadDeadline(time.Now().Add(10 * time.Millisecond))

    header := make([]byte, 4)
    n, err := io.ReadFull(conn, header)
    if err != nil {
        if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
            return nil, nil // No data available, return nil
        }
        if err == io.EOF {
            return nil, fmt.Errorf("socket disconnected")
        }
        return nil, err
    }

    if n == 0 {
        return nil, fmt.Errorf("socket disconnected")
    }

    logMsg(threadName, fmt.Sprintf("Received header %s", asHexStream(header)))

    if n != 4 {
        return nil, fmt.Errorf("received partial header (implementation does not support partial receives yet)")
    }

    prefix := header[0:2]
    if !bytes.Equal(prefix, HEADER_PREFIX) {
        return nil, fmt.Errorf("framing error: header prefix %s != %s", asHexStream(prefix), asHexStream(HEADER_PREFIX))
    }

    size := binary.BigEndian.Uint16(header[2:4])
    logMsg(threadName, fmt.Sprintf("Trying to read %d bytes datagram", size))

    // Remove timeout for reading the full datagram
    conn.SetReadDeadline(time.Time{})
    datagram := make([]byte, size)
    n, err = io.ReadFull(conn, datagram)
    if err != nil {
        return nil, err
    }

    logMsg(threadName, fmt.Sprintf("Received datagram %s", asHexStream(datagram)))

    if n != int(size) {
        return nil, fmt.Errorf("received partial datagram (implementation does not support partial receives yet)")
    }

    return datagram, nil
}

func handleKeepalive(conn net.Conn, threadName string) {
    logMsg(threadName, "Keep alive received")
}

func handleBye(conn net.Conn, datagram []byte, threadName string) {
    logMsg(threadName, "Bye received")
    reason := datagram[1:]
    logMsg(threadName, fmt.Sprintf("Bye reason: %s", string(reason)))
}

func handlePayloadWithIdentifier(conn net.Conn, datagram []byte, payloadReceivedCallback func(string, byte, int64, []byte), threadName string) {
    logMsg(threadName, "Payload with identifier received")
    identifier := string(datagram[1:9])
    payloadType := datagram[9]
    originTimestamp := int64(binary.BigEndian.Uint64(datagram[10:18]))
    payload := datagram[18:]

    logMsg(threadName, fmt.Sprintf("Payload received (identifier: %s, payload_type: %s, origin_timestamp: %d): %s", 
        identifier, asHexStream([]byte{payloadType}), originTimestamp, asHexStream(payload)))

    payloadReceivedCallback(identifier, payloadType, originTimestamp, payload)
}

func handleTimestampRequest(conn net.Conn, datagram []byte, threadName string) {
    logMsg(threadName, "Timestamp request received")
    timestampT0 := int64(binary.BigEndian.Uint64(datagram[1:]))
    timestampT1 := currentTimestamp()
    logMsg(threadName, fmt.Sprintf("Timestamp request delta: %dms", timestampT1-timestampT0))
    writeTimestampResponse(conn, timestampT0, timestampT1, threadName)
}

func handleDatagram(conn net.Conn, datagram []byte, payloadReceivedCallback func(string, byte, int64, []byte), threadName string) {
    datagramType := datagram[0]

    switch datagramType {
    case 0x00:
        handleKeepalive(conn, threadName)
    case 0x02:
        handleBye(conn, datagram, threadName)
    case 0x05:
        handlePayloadWithIdentifier(conn, datagram, payloadReceivedCallback, threadName)
    case 0x06:
        handleTimestampRequest(conn, datagram, threadName)
    default:
        logMsg(threadName, fmt.Sprintf("Unknown/unimplemented datagram type %s received", asHexStream([]byte{datagramType})))
    }
}

func runStreamingClient(host string, port int, sessionToken string, useTLS bool, payloadReceivedCallback func(string, byte, int64, []byte), loopCallback func(net.Conn), threadName string) error {
    conn, err := connect(host, port, useTLS, threadName)
    if err != nil {
        return err
    }
    defer conn.Close()

    if err := handshake(conn, threadName); err != nil {
        return err
    }

    if err := writeToken(conn, sessionToken, threadName); err != nil {
        return err
    }

    for {
        datagram, err := readDatagram(conn, threadName)
        if err != nil {
            return err
        }

        if datagram != nil {
            handleDatagram(conn, datagram, payloadReceivedCallback, threadName)
        } else {
            time.Sleep(10 * time.Millisecond)
        }

        loopCallback(conn)
    }
}

private static Socket connect(String host, int port, boolean tls, String threadName) throws Exception {
    Socket socket;

    if (tls) {
        SSLSocketFactory sslSocketFactory = (SSLSocketFactory) SSLSocketFactory.getDefault();
        SSLSocket sslSocket = (SSLSocket) sslSocketFactory.createSocket(host, port);
        sslSocket.startHandshake();

        log(threadName, String.format("Connected to %s:%d (TLS: %b)", host, port, tls));

        SSLSession session = sslSocket.getSession();
        log(threadName, String.format("TLS handshake successful - Protocol: %s, Cipher: %s", 
            session.getProtocol(), session.getCipherSuite()));

        socket = sslSocket;
    } else {
        socket = new Socket(host, port);
        log(threadName, String.format("Connected to %s:%d (TLS: %b)", host, port, tls));
    }

    return socket;
}

private static void handshake(Socket socket, String threadName) throws Exception {
    byte[] protocolVersionByte = {0x01};
    socket.getOutputStream().write(protocolVersionByte);
    socket.getOutputStream().flush();

    byte[] recv = new byte[1];
    int bytesRead = socket.getInputStream().read(recv);
    if (bytesRead != 1) {
        throw new Exception("Failed to receive protocol version");
    }

    log(threadName, "Received protocol version " + asHexStream(recv));

    if (recv[0] != 1) {
        throw new Exception("Unsupported protocol version received");
    }
}

private static void writeDatagram(Socket socket, byte[] datagram, String threadName) throws Exception {
    ByteBuffer buffer = ByteBuffer.allocate(4 + datagram.length);
    buffer.put(HEADER_PREFIX);
    buffer.putShort((short) datagram.length);
    buffer.put(datagram);

    byte[] frame = buffer.array();
    log(threadName, "Writing frame " + asHexStream(frame));

    socket.getOutputStream().write(frame);
    socket.getOutputStream().flush();
}

private static void writeToken(Socket socket, String token, String threadName) throws Exception {
    log(threadName, "Writing token " + token);
    ByteBuffer datagram = ByteBuffer.allocate(1 + token.length());
    datagram.put((byte) 0x01);
    datagram.put(token.getBytes(StandardCharsets.US_ASCII));
    writeDatagram(socket, datagram.array(), threadName);
}

private static void writeKeepalive(Socket socket, String threadName) throws Exception {
    log(threadName, "Writing keep alive");
    byte[] datagram = {0x00};
    writeDatagram(socket, datagram, threadName);
}

private static void writeTimestampResponse(Socket socket, long timestampT0, long timestampT1, String threadName) throws Exception {
    long timestampT2 = currentTimestamp();
    log(threadName, String.format("Writing timestamp response (t0: %d, t1: %d, t2: %d)", timestampT0, timestampT1, timestampT2));

    ByteBuffer datagram = ByteBuffer.allocate(25);
    datagram.put((byte) 0x07);
    datagram.putLong(timestampT0);
    datagram.putLong(timestampT1);
    datagram.putLong(timestampT2);
    writeDatagram(socket, datagram.array(), threadName);
}

private static void writePayloadWithIdentifier(Socket socket, String identifier, byte payloadType, 
        byte[] payload, String threadName) throws Exception {
    log(threadName, String.format("Writing payload with identifier (identifier: %s, payload_type: %s): %s",
        identifier, asHexStream(new byte[]{payloadType}), asHexStream(payload)));

    ByteBuffer datagram = ByteBuffer.allocate(1 + 8 + 1 + 8 + payload.length);
    datagram.put((byte) 0x05);

    // Pad identifier to 8 bytes
    byte[] identifierBytes = identifier.getBytes(StandardCharsets.US_ASCII);
    byte[] paddedIdentifier = new byte[8];
    System.arraycopy(identifierBytes, 0, paddedIdentifier, 0, Math.min(identifierBytes.length, 8));
    datagram.put(paddedIdentifier);

    datagram.put(payloadType);
    datagram.putLong(currentTimestamp());
    datagram.put(payload);

    writeDatagram(socket, datagram.array(), threadName);
}

private static byte[] readDatagram(Socket socket, String threadName) throws Exception {
    InputStream is = socket.getInputStream();

    // Try non-blocking read
    socket.setSoTimeout(10);

    byte[] header = new byte[4];
    try {
        int bytesRead = 0;
        while (bytesRead < 4) {
            int n = is.read(header, bytesRead, 4 - bytesRead);
            if (n == -1) {
                throw new Exception("Socket disconnected");
            }
            bytesRead += n;
        }
    } catch (SocketTimeoutException e) {
        return null; // No data available
    }

    log(threadName, "Received header " + asHexStream(header));

    if (header[0] != HEADER_PREFIX[0] || header[1] != HEADER_PREFIX[1]) {
        throw new Exception(String.format("Framing error: header prefix %s != %s", 
            asHexStream(new byte[]{header[0], header[1]}), asHexStream(HEADER_PREFIX)));
    }

    int size = ((header[2] & 0xFF) << 8) | (header[3] & 0xFF);
    log(threadName, String.format("Trying to read %d bytes datagram", size));

    // Set blocking for reading the full datagram
    socket.setSoTimeout(0);

    byte[] datagram = new byte[size];
    int bytesRead = 0;
    while (bytesRead < size) {
        int n = is.read(datagram, bytesRead, size - bytesRead);
        if (n == -1) {
            throw new Exception("Socket disconnected");
        }
        bytesRead += n;
    }

    log(threadName, "Received datagram " + asHexStream(datagram));
    return datagram;
}

private static void handleKeepalive(Socket socket, String threadName) {
    log(threadName, "Keep alive received");
}

private static void handleBye(Socket socket, byte[] datagram, String threadName) {
    log(threadName, "Bye received");
    if (datagram.length > 1) {
        String reason = new String(datagram, 1, datagram.length - 1, StandardCharsets.UTF_8);
        log(threadName, "Bye reason: " + reason);
    }
}

@FunctionalInterface
interface PayloadCallback {
    void onPayload(String identifier, byte payloadType, long originTimestamp, byte[] payload);
}

private static void handlePayloadWithIdentifier(Socket socket, byte[] datagram, 
        PayloadCallback callback, String threadName) {
    log(threadName, "Payload with identifier received");

    String identifier = new String(datagram, 1, 8, StandardCharsets.US_ASCII).trim();
    byte payloadType = datagram[9];
    long originTimestamp = ByteBuffer.wrap(datagram, 10, 8).getLong();
    byte[] payload = new byte[datagram.length - 18];
    System.arraycopy(datagram, 18, payload, 0, payload.length);

    log(threadName, String.format("Payload received (identifier: %s, payload_type: %s, origin_timestamp: %d): %s",
        identifier, asHexStream(new byte[]{payloadType}), originTimestamp, asHexStream(payload)));

    callback.onPayload(identifier, payloadType, originTimestamp, payload);
}

private static void handleTimestampRequest(Socket socket, byte[] datagram, String threadName) throws Exception {
    log(threadName, "Timestamp request received");
    long timestampT0 = ByteBuffer.wrap(datagram, 1, 8).getLong();
    long timestampT1 = currentTimestamp();
    log(threadName, String.format("Timestamp request delta: %dms", timestampT1 - timestampT0));
    writeTimestampResponse(socket, timestampT0, timestampT1, threadName);
}

private static void handleDatagram(Socket socket, byte[] datagram, PayloadCallback callback, String threadName) throws Exception {
    byte datagramType = datagram[0];

    switch (datagramType) {
        case 0x00:
            handleKeepalive(socket, threadName);
            break;
        case 0x02:
            handleBye(socket, datagram, threadName);
            break;
        case 0x05:
            handlePayloadWithIdentifier(socket, datagram, callback, threadName);
            break;
        case 0x06:
            handleTimestampRequest(socket, datagram, threadName);
            break;
        default:
            log(threadName, String.format("Unknown/unimplemented datagram type %s received", 
                asHexStream(new byte[]{datagramType})));
    }
}

@FunctionalInterface
interface LoopCallback {
    void onLoop(Socket socket) throws Exception;
}

private static void runStreamingClient(String host, int port, String sessionToken, boolean tls,
        PayloadCallback payloadCallback, LoopCallback loopCallback, String threadName) throws Exception {
    Socket socket = connect(host, port, tls, threadName);

    try {
        handshake(socket, threadName);
        writeToken(socket, sessionToken, threadName);

        while (!socket.isClosed()) {
            byte[] datagram = readDatagram(socket, threadName);
            if (datagram != null) {
                handleDatagram(socket, datagram, payloadCallback, threadName);
            } else {
                Thread.sleep(10);
            }

            loopCallback.onLoop(socket);
        }
    } finally {
        socket.close();
    }
}

enum Connection {
    Plain(TcpStream),
    Tls(TlsStream<TcpStream>),
}

impl Connection {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        match self {
            Connection::Plain(stream) => stream.read(buf),
            Connection::Tls(stream) => stream.read(buf),
        }
    }

    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        match self {
            Connection::Plain(stream) => stream.write(buf),
            Connection::Tls(stream) => stream.write(buf),
        }
    }

    fn set_nonblocking(&self, nonblocking: bool) -> std::io::Result<()> {
        match self {
            Connection::Plain(stream) => stream.set_nonblocking(nonblocking),
            Connection::Tls(stream) => stream.get_ref().set_nonblocking(nonblocking),
        }
    }
}

fn connect(host: &str, port: u16, use_tls: bool, thread_name: &str) -> Result<Connection, Box<dyn std::error::Error + Send + Sync>> {
    let stream = TcpStream::connect(format!("{}:{}", host, port))?;

    if use_tls {
        let connector = TlsConnector::new()?;
        let tls_stream = connector.connect(host, stream)?;

        log_msg(thread_name, &format!("Connected to {}:{} (TLS: {})", host, port, use_tls));
        log_msg(thread_name, "TLS handshake successful");

        Ok(Connection::Tls(tls_stream))
    } else {
        log_msg(thread_name, &format!("Connected to {}:{} (TLS: {})", host, port, use_tls));
        Ok(Connection::Plain(stream))
    }
}

fn handshake(conn: &mut Connection, thread_name: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let protocol_version_byte = [0x01];
    conn.write(&protocol_version_byte)?;

    let mut recv = [0u8; 1];
    conn.read(&mut recv)?;

    log_msg(thread_name, &format!("Received protocol version {}", as_hex_stream(&recv)));

    if recv[0] != 1 {
        return Err("Unsupported protocol version received".into());
    }

    Ok(())
}

fn write_datagram(conn: &mut Connection, datagram: &[u8], thread_name: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let mut header = Vec::new();
    header.extend_from_slice(&HEADER_PREFIX);
    header.extend_from_slice(&(datagram.len() as u16).to_be_bytes());

    let mut frame = header;
    frame.extend_from_slice(datagram);

    log_msg(thread_name, &format!("Writing frame {}", as_hex_stream(&frame)));

    conn.write(&frame)?;
    Ok(())
}

fn write_token(conn: &mut Connection, token: &str, thread_name: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    log_msg(thread_name, &format!("Writing token {}", token));
    let mut datagram = vec![0x01];
    datagram.extend_from_slice(token.as_bytes());
    write_datagram(conn, &datagram, thread_name)
}

fn write_keepalive(conn: &mut Connection, thread_name: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    log_msg(thread_name, "Writing keep alive");
    let datagram = [0x00];
    write_datagram(conn, &datagram, thread_name)
}

fn write_timestamp_response(
    conn: &mut Connection,
    timestamp_t0: u64,
    timestamp_t1: u64,
    thread_name: &str,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let timestamp_t2 = current_timestamp();
    log_msg(
        thread_name,
        &format!("Writing timestamp response (t0: {}, t1: {}, t2: {})", timestamp_t0, timestamp_t1, timestamp_t2),
    );

    let mut datagram = vec![0x07];
    datagram.extend_from_slice(&timestamp_t0.to_be_bytes());
    datagram.extend_from_slice(&timestamp_t1.to_be_bytes());
    datagram.extend_from_slice(&timestamp_t2.to_be_bytes());

    write_datagram(conn, &datagram, thread_name)
}

fn write_payload_with_identifier(
    conn: &mut Connection,
    identifier: &str,
    payload_type: u8,
    payload: &[u8],
    thread_name: &str,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    log_msg(
        thread_name,
        &format!(
            "Writing payload with identifier (identifier: {}, payload_type: {}): {}",
            identifier,
            as_hex_stream(&[payload_type]),
            as_hex_stream(payload)
        ),
    );

    let mut datagram = vec![0x05];
    datagram.extend_from_slice(identifier.as_bytes());
    datagram.push(payload_type);
    datagram.extend_from_slice(&current_timestamp().to_be_bytes());
    datagram.extend_from_slice(payload);

    write_datagram(conn, &datagram, thread_name)
}

fn read_datagram(conn: &mut Connection, thread_name: &str) -> Result<Option<Vec<u8>>, Box<dyn std::error::Error + Send + Sync>> {
    conn.set_nonblocking(true)?;

    let mut header = [0u8; 4];
    match conn.read(&mut header) {
        Ok(0) => return Err("Socket disconnected".into()),
        Ok(n) if n != 4 => return Err("Received partial header (implementation does not support partial receives yet)".into()),
        Ok(_) => {},
        Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => return Ok(None),
        Err(e) => return Err(e.into()),
    }

    log_msg(thread_name, &format!("Received header {}", as_hex_stream(&header)));

    if header[0..2] != HEADER_PREFIX {
        return Err(format!("Framing error: header prefix {} != {}", 
                          as_hex_stream(&header[0..2]), 
                          as_hex_stream(&HEADER_PREFIX)).into());
    }

    let size = u16::from_be_bytes([header[2], header[3]]) as usize;
    log_msg(thread_name, &format!("Trying to read {} bytes datagram", size));

    conn.set_nonblocking(false)?;
    let mut datagram = vec![0u8; size];
    conn.read(&mut datagram)?;

    log_msg(thread_name, &format!("Received datagram {}", as_hex_stream(&datagram)));

    Ok(Some(datagram))
}

fn handle_keepalive(_conn: &mut Connection, thread_name: &str) {
    log_msg(thread_name, "Keep alive received");
}

fn handle_bye(_conn: &mut Connection, datagram: &[u8], thread_name: &str) {
    log_msg(thread_name, "Bye received");
    let reason = String::from_utf8_lossy(&datagram[1..]);
    log_msg(thread_name, &format!("Bye reason: {}", reason));
}

fn handle_payload_with_identifier(
    _conn: &mut Connection,
    datagram: &[u8],
    payload_received_callback: &dyn Fn(&str, u8, u64, &[u8]),
    thread_name: &str,
) {
    log_msg(thread_name, "Payload with identifier received");
    let identifier = String::from_utf8_lossy(&datagram[1..9]);
    let payload_type = datagram[9];
    let origin_timestamp = u64::from_be_bytes([
        datagram[10], datagram[11], datagram[12], datagram[13],
        datagram[14], datagram[15], datagram[16], datagram[17],
    ]);
    let payload = &datagram[18..];

    log_msg(
        thread_name,
        &format!(
            "Payload received (identifier: {}, payload_type: {}, origin_timestamp: {}): {}",
            identifier,
            as_hex_stream(&[payload_type]),
            origin_timestamp,
            as_hex_stream(payload)
        ),
    );

    payload_received_callback(&identifier, payload_type, origin_timestamp, payload);
}

fn handle_timestamp_request(conn: &mut Connection, datagram: &[u8], thread_name: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    log_msg(thread_name, "Timestamp request received");
    let timestamp_t0 = u64::from_be_bytes([
        datagram[1], datagram[2], datagram[3], datagram[4],
        datagram[5], datagram[6], datagram[7], datagram[8],
    ]);
    let timestamp_t1 = current_timestamp();
    log_msg(thread_name, &format!("Timestamp request delta: {}ms", timestamp_t1 - timestamp_t0));
    write_timestamp_response(conn, timestamp_t0, timestamp_t1, thread_name)
}

fn handle_datagram(
    conn: &mut Connection,
    datagram: &[u8],
    payload_received_callback: &dyn Fn(&str, u8, u64, &[u8]),
    thread_name: &str,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    if datagram.is_empty() {
        return Ok(());
    }

    match datagram[0] {
        0x00 => handle_keepalive(conn, thread_name),
        0x02 => handle_bye(conn, datagram, thread_name),
        0x05 => handle_payload_with_identifier(conn, datagram, payload_received_callback, thread_name),
        0x06 => handle_timestamp_request(conn, datagram, thread_name)?,
        _ => log_msg(thread_name, &format!("Unknown/unimplemented datagram type {} received", as_hex_stream(&[datagram[0]]))),
    }

    Ok(())
}

fn run_streaming_client(
    host: &str,
    port: u16,
    session_token: &str,
    use_tls: bool,
    payload_received_callback: Box<dyn Fn(&str, u8, u64, &[u8]) + Send>,
    loop_callback: Box<dyn Fn(&mut Connection) + Send>,
    thread_name: &str,
    shutdown_flag: Arc<AtomicBool>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let mut conn = connect(host, port, use_tls, thread_name)?;

    handshake(&mut conn, thread_name)?;
    write_token(&mut conn, session_token, thread_name)?;

    while !shutdown_flag.load(Ordering::Relaxed) {
        match read_datagram(&mut conn, thread_name)? {
            Some(datagram) => {
                handle_datagram(&mut conn, &datagram, &*payload_received_callback, thread_name)?;
            }
            None => {
                thread::sleep(Duration::from_millis(10));
            }
        }

        loop_callback(&mut conn);
    }

    log_msg(thread_name, "Shutdown signal received, terminating");
    Ok(())
}

public abstract class Connection : IDisposable
{
    public abstract int Read(byte[] buffer, int offset, int count);
    public abstract void Write(byte[] buffer, int offset, int count);
    public abstract void SetNonBlocking(bool nonBlocking);
    public abstract void Close();
    public abstract void Dispose();
}

public class PlainConnection : Connection
{
    private readonly TcpClient tcpClient;
    private readonly NetworkStream stream;

    public PlainConnection(TcpClient tcpClient)
    {
        this.tcpClient = tcpClient;
        this.stream = tcpClient.GetStream();
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        return stream.Read(buffer, offset, count);
    }

    public override void Write(byte[] buffer, int offset, int count)
    {
        stream.Write(buffer, offset, count);
    }

    public override void SetNonBlocking(bool nonBlocking)
    {
        tcpClient.Client.Blocking = !nonBlocking;
    }

    public override void Close()
    {
        stream.Close();
        tcpClient.Close();
    }

    public override void Dispose()
    {
        stream.Dispose();
        tcpClient.Dispose();
    }
}

public class TlsConnection : Connection
{
    private readonly TcpClient tcpClient;
    private readonly SslStream sslStream;

    public TlsConnection(TcpClient tcpClient, string targetHost)
    {
        this.tcpClient = tcpClient;
        this.sslStream = new SslStream(tcpClient.GetStream(), false, ValidateServerCertificate);
        this.sslStream.AuthenticateAsClient(targetHost);
    }

    private static bool ValidateServerCertificate(object sender, X509Certificate? certificate, X509Chain? chain, SslPolicyErrors sslPolicyErrors)
    {
        // In production, implement proper certificate validation
        return true;
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        return sslStream.Read(buffer, offset, count);
    }

    public override void Write(byte[] buffer, int offset, int count)
    {
        sslStream.Write(buffer, offset, count);
    }

    public override void SetNonBlocking(bool nonBlocking)
    {
        tcpClient.Client.Blocking = !nonBlocking;
    }

    public override void Close()
    {
        sslStream.Close();
        tcpClient.Close();
    }

    public override void Dispose()
    {
        sslStream.Dispose();
        tcpClient.Dispose();
    }
}

public static class TcpStreaming
{
    public static Connection Connect(string host, int port, bool useTls, string threadName)
    {
        var tcpClient = new TcpClient(host, port);

        if (useTls)
        {
            var tlsConnection = new TlsConnection(tcpClient, host);
            Utilities.LogMessage(threadName, $"Connected to {host}:{port} (TLS: {useTls})");
            Utilities.LogMessage(threadName, "TLS handshake successful");
            return tlsConnection;
        }
        else
        {
            Utilities.LogMessage(threadName, $"Connected to {host}:{port} (TLS: {useTls})");
            return new PlainConnection(tcpClient);
        }
    }

    public static void Handshake(Connection conn, string threadName)
    {
        var protocolVersionByte = new byte[] { 0x01 };
        conn.Write(protocolVersionByte, 0, protocolVersionByte.Length);

        var recv = new byte[1];
        conn.Read(recv, 0, recv.Length);

        Utilities.LogMessage(threadName, $"Received protocol version {Utilities.AsHexStream(recv)}");

        if (recv[0] != 1)
        {
            throw new InvalidOperationException("Unsupported protocol version received");
        }
    }

    public static void WriteDatagram(Connection conn, byte[] datagram, string threadName)
    {
        var header = new List<byte>();
        header.AddRange(Constants.HeaderPrefix);
        header.AddRange(BitConverter.GetBytes((ushort)datagram.Length).Reverse()); // Big-endian

        var frame = new List<byte>();
        frame.AddRange(header);
        frame.AddRange(datagram);

        var frameArray = frame.ToArray();
        Utilities.LogMessage(threadName, $"Writing frame {Utilities.AsHexStream(frameArray)}");

        conn.Write(frameArray, 0, frameArray.Length);
    }

    public static void WriteToken(Connection conn, string token, string threadName)
    {
        Utilities.LogMessage(threadName, $"Writing token {token}");
        var datagram = new List<byte> { 0x01 };
        datagram.AddRange(Encoding.UTF8.GetBytes(token));
        WriteDatagram(conn, datagram.ToArray(), threadName);
    }

    public static void WriteKeepAlive(Connection conn, string threadName)
    {
        Utilities.LogMessage(threadName, "Writing keep alive");
        var datagram = new byte[] { 0x00 };
        WriteDatagram(conn, datagram, threadName);
    }

    public static void WriteTimestampResponse(Connection conn, long timestampT0, long timestampT1, string threadName)
    {
        var timestampT2 = Utilities.CurrentTimestamp();
        Utilities.LogMessage(threadName, $"Writing timestamp response (t0: {timestampT0}, t1: {timestampT1}, t2: {timestampT2})");

        var datagram = new List<byte> { 0x07 };
        datagram.AddRange(BitConverter.GetBytes(timestampT0).Reverse()); // Big-endian
        datagram.AddRange(BitConverter.GetBytes(timestampT1).Reverse());
        datagram.AddRange(BitConverter.GetBytes(timestampT2).Reverse());

        WriteDatagram(conn, datagram.ToArray(), threadName);
    }

    public static void WritePayloadWithIdentifier(Connection conn, string identifier, byte payloadType, byte[] payload, string threadName)
    {
        Utilities.LogMessage(threadName, 
            $"Writing payload with identifier (identifier: {identifier}, payload_type: {Utilities.AsHexStream(new[] { payloadType })}): {Utilities.AsHexStream(payload)}");

        var datagram = new List<byte> { 0x05 };
        datagram.AddRange(Encoding.UTF8.GetBytes(identifier));
        datagram.Add(payloadType);
        datagram.AddRange(BitConverter.GetBytes(Utilities.CurrentTimestamp()).Reverse()); // Big-endian
        datagram.AddRange(payload);

        WriteDatagram(conn, datagram.ToArray(), threadName);
    }

    public static byte[]? ReadDatagram(Connection conn, string threadName)
    {
        // Set non-blocking mode for header read
        conn.SetNonBlocking(true);

        var header = new byte[4];
        int headerBytesRead = 0;

        try
        {
            headerBytesRead = conn.Read(header, 0, 4);
        }
        catch (SocketException ex) when (ex.SocketErrorCode == SocketError.WouldBlock)
        {
            // Socket would block - no data available
            return null;
        }
        catch (IOException ex) when (ex.InnerException is SocketException socketEx && 
                                     socketEx.SocketErrorCode == SocketError.WouldBlock)
        {
            // SSL stream wraps socket exception in IOException
            return null;
        }

        if (headerBytesRead == 0)
        {
            throw new InvalidOperationException("Socket disconnected");
        }

        if (headerBytesRead != 4)
        {
            throw new InvalidOperationException("Received partial header (implementation does not support partial receives yet)");
        }

        Utilities.LogMessage(threadName, $"Received header {Utilities.AsHexStream(header)}");

        if (!header.Take(2).SequenceEqual(Constants.HeaderPrefix))
        {
            throw new InvalidOperationException($"Framing error: header prefix {Utilities.AsHexStream(header.Take(2).ToArray())} != {Utilities.AsHexStream(Constants.HeaderPrefix)}");
        }

        var size = (ushort)((header[2] << 8) | header[3]); // Big-endian
        Utilities.LogMessage(threadName, $"Trying to read {size} bytes datagram");

        // Set blocking mode for datagram body read (to ensure complete read)
        conn.SetNonBlocking(false);

        var datagram = new byte[size];
        var datagramBytesRead = 0;
        while (datagramBytesRead < size)
        {
            var bytesRead = conn.Read(datagram, datagramBytesRead, size - datagramBytesRead);
            if (bytesRead == 0)
                throw new InvalidOperationException("Socket disconnected");
            datagramBytesRead += bytesRead;
        }

        Utilities.LogMessage(threadName, $"Received datagram {Utilities.AsHexStream(datagram)}");

        return datagram;
    }

    public static void HandleKeepAlive(Connection conn, string threadName)
    {
        Utilities.LogMessage(threadName, "Keep alive received");
    }

    public static void HandleBye(Connection conn, byte[] datagram, string threadName)
    {
        Utilities.LogMessage(threadName, "Bye received");
        var reason = Encoding.UTF8.GetString(datagram, 1, datagram.Length - 1);
        Utilities.LogMessage(threadName, $"Bye reason: {reason}");
    }

    public static void HandlePayloadWithIdentifier(Connection conn, byte[] datagram, Action<string, byte, long, byte[]> payloadReceivedCallback, string threadName)
    {
        Utilities.LogMessage(threadName, "Payload with identifier received");
        var identifier = Encoding.UTF8.GetString(datagram, 1, 8);
        var payloadType = datagram[9];
        var originTimestamp = BitConverter.ToInt64(datagram.Skip(10).Take(8).Reverse().ToArray(), 0);
        var payload = datagram.Skip(18).ToArray();

        Utilities.LogMessage(threadName, 
            $"Payload received (identifier: {identifier}, payload_type: {Utilities.AsHexStream(new[] { payloadType })}, origin_timestamp: {originTimestamp}): {Utilities.AsHexStream(payload)}");

        payloadReceivedCallback(identifier, payloadType, originTimestamp, payload);
    }

    public static void HandleTimestampRequest(Connection conn, byte[] datagram, string threadName)
    {
        Utilities.LogMessage(threadName, "Timestamp request received");
        var timestampT0 = BitConverter.ToInt64(datagram.Skip(1).Take(8).Reverse().ToArray(), 0);
        var timestampT1 = Utilities.CurrentTimestamp();
        Utilities.LogMessage(threadName, $"Timestamp request delta: {timestampT1 - timestampT0}ms");
        WriteTimestampResponse(conn, timestampT0, timestampT1, threadName);
    }

    public static void HandleDatagram(Connection conn, byte[] datagram, Action<string, byte, long, byte[]> payloadReceivedCallback, string threadName)
    {
        if (datagram.Length == 0) return;

        switch (datagram[0])
        {
            case 0x00:
                HandleKeepAlive(conn, threadName);
                break;
            case 0x02:
                HandleBye(conn, datagram, threadName);
                break;
            case 0x05:
                HandlePayloadWithIdentifier(conn, datagram, payloadReceivedCallback, threadName);
                break;
            case 0x06:
                HandleTimestampRequest(conn, datagram, threadName);
                break;
            default:
                Utilities.LogMessage(threadName, $"Unknown/unimplemented datagram type {Utilities.AsHexStream(new[] { datagram[0] })} received");
                break;
        }
    }

    public static void RunStreamingClient(
        string host, 
        int port, 
        string sessionToken, 
        bool useTls, 
        Action<string, byte, long, byte[]> payloadReceivedCallback, 
        Action<Connection> loopCallback, 
        string threadName,
        CancellationToken cancellationToken)
    {
        using var conn = Connect(host, port, useTls, threadName);

        Handshake(conn, threadName);
        WriteToken(conn, sessionToken, threadName);

        while (!cancellationToken.IsCancellationRequested)
        {
            var datagram = ReadDatagram(conn, threadName);
            if (datagram != null)
            {
                HandleDatagram(conn, datagram, payloadReceivedCallback, threadName);
            }
            else
            {
                Thread.Sleep(10);
            }

            loopCallback(conn);
        }

        Utilities.LogMessage(threadName, "Shutdown signal received, terminating");
    }
}

Explanation

PythonGoJavaRust.NET