Dial

Used for raw Socket access, supporting tcp, udp, tls, unix protocols. Supports 4 popular communication protocols: mqtt, nats, amqp, kafka. Supports database connections, including: sqlite3, mysql, postgres, clickhouse.

    
    
    
Dial(address)

    
    
    
Dial(address, timeout)

    
    
    
Dial(address, options)

Examples

• Dial function call example:

  • javascript
  • python
  • cpp

  
      javascript
      
      
  function main(){
      // Dial supports tcp://, udp://, tls://, unix:// protocols, can add a parameter to specify timeout in seconds
      var client = Dial("tls://www.baidu.com:443")
      if (client) {
          // write can append a numeric parameter to specify timeout, returns the number of bytes successfully sent
          client.write("GET / HTTP/1.1\nConnection: Closed\n\n")
          while (true) {
              // read can append a numeric parameter to specify timeout, unit: milliseconds. Returns null indicating error, timeout, or socket closed
              var buf = client.read()
              if (!buf) {
                   break
              }
              Log(buf)
          }
          client.close()
      }
  }
  

  
      python
      
      
  def main():
      client = Dial("tls://www.baidu.com:443")
      if client:
          client.write("GET / HTTP/1.1\nConnection: Closed\n\n")
          while True:
              buf = client.read()
              if not buf:
                  break
              Log(buf)
          client.close()
  

  
      c++
      
      
  void main() {
      auto client = Dial("tls://www.baidu.com:443");
      if(client.Valid) {
          client.write("GET / HTTP/1.1\nConnection: Closed\n\n");
          while(true) {
              auto buf = client.read();
              if(buf == "") {
                  break;
              }
              Log(buf);
          }
          client.close();
      }
  }
  

• Accessing Binance WebSocket market data interface:

  • javascript
  • python
  • cpp

  
      javascript
      
      
  function main() {
      LogStatus("Connecting...")
      // Access Binance WebSocket interface
      var client = Dial("wss://stream.binance.com:9443/ws/!ticker@arr")
      if (!client) {
          Log("Connection failed, exiting")
          return
      }
  
      while (true) {
          // read only returns data obtained after calling read
          var buf = client.read()
          if (!buf) {
              break
          }
          var table = {
              type: 'table',
              title: 'Market Chart',
              cols: ['Symbol', 'High', 'Low', 'Bid', 'Ask', 'Last Price', 'Volume', 'Update Time'],
              rows: []
          }
          var obj = JSON.parse(buf)
          _.each(obj, function(ticker) {
              table.rows.push([ticker.s, ticker.h, ticker.l, ticker.b, ticker.a, ticker.c, ticker.q, _D(ticker.E)])
          })
          LogStatus('`' + JSON.stringify(table) + '`')
      }
      client.close()
  }
  

  
      python
      
      
  import json
  def main():
      LogStatus("Connecting...")
      client = Dial("wss://stream.binance.com:9443/ws/!ticker@arr")
      if not client:
          Log("Connection failed, exiting")
          return
  
      while True:
          buf = client.read()
          if not buf:
              break
          table = {
              "type" : "table",
              "title" : "Market Chart",
              "cols" : ["Symbol", "High", "Low", "Bid", "Ask", "Last Price", "Volume", "Update Time"],
              "rows" : []
          }
          obj = json.loads(buf)
          for i in range(len(obj)):
              table["rows"].append([obj[i]["s"], obj[i]["h"], obj[i]["l"], obj[i]["b"], obj[i]["a"], obj[i]["c"], obj[i]["q"], _D(int(obj[i]["E"]))])
          LogStatus('`' + json.dumps(table) + '`')
      client.close()
  

  
      c++
      
      
  void main() {
      LogStatus("Connecting...");
      auto client = Dial("wss://stream.binance.com:9443/ws/!ticker@arr");
      if(!client.Valid) {
          Log("Connection failed, exiting");
          return;
      }
  
      while(true) {
          auto buf = client.read();
          if(buf == "") {
              break;
          }
          json table = R"({
              "type" : "table",
              "title" : "Market Chart",
              "cols" : ["Symbol", "High", "Low", "Bid", "Ask", "Last Price", "Volume", "Update Time"],
              "rows" : []
          })"_json;
          json obj = json::parse(buf);
          for(auto& ele : obj.items()) {
              table["rows"].push_back({ele.value()["s"], ele.value()["h"], ele.value()["l"], ele.value()["b"], ele.value()["a"], ele.value()["c"],
                  ele.value()["q"], _D(ele.value()["E"])});
          }
          LogStatus("`" + table.dump() + "`");
      }
      client.close();
  }
  

• Access Binance's WebSocket interface and set wss request headers.

  • javascript
  • python
  • cpp

  
      javascript
      
      
  function main() {
      let options = {"headers": {"X-MBX-APIKEY": "your access key"}}
      let random = `fmz${UnixNano()}`
      let ts = new Date().getTime()
      let secretKey = "your secret key"
      let topic = "com_announcement_en"
      let payload = `random=${random}&topic=${topic}&recvWindow=30000&timestamp=${ts}`
      let signature = Encode("sha256", "string", "hex", payload, "string", secretKey)
      let query = `?${payload}&signature=${signature}`
  
      Log("query:", query)
      let conn = Dial(`wss://api.binance.com/sapi/wss${query}`, options)
  
      for (var i = 0 ; i < 10 ; i++) {
          let ret = conn.read()
          Log(ret)
      }
  }
  

  
      python
      
      
  import time
  
  def main():
      options = {"headers": {"X-MBX-APIKEY": "your access key"}}
      random = "fmz" + str(UnixNano())
      ts = int(time.time() * 1000)
      secretKey = "your secret key"
      topic = "com_announcement_en"
      payload = f"random={random}&topic={topic}&recvWindow=30000&timestamp={ts}"
      signature = Encode("sha256", "string", "hex", payload, "string", secretKey)
      query = f"?{payload}&signature={signature}"
  
      Log("query:", query)
      conn = Dial(f"wss://api.binance.com/sapi/wss{query}", options)
  
      for i in range(10):
          ret = conn.read()
          Log(ret)
  

  
      c++
      
      
  // Not supported currently

• Accessing OKX WebSocket market data interface:

  • javascript
  • python
  • cpp

  
      javascript
      
      
  var ws = null
  function main(){
      var param = {
          "op": "subscribe",
          "args": [{
              "channel": "tickers",
              "instId": "BTC-USDT"
          }]
      }
      // When calling the Dial function, specify reconnect=true to set reconnection mode, and specify payload as the message to send upon reconnection. After the WebSocket connection is disconnected, it will automatically reconnect and automatically send the message
      ws = Dial("wss://ws.okx.com:8443/ws/v5/public|compress=gzip_raw&mode=recv&reconnect=true&payload=" + JSON.stringify(param))
      if(ws){
          var pingCyc = 1000 * 20
          var lastPingTime = new Date().getTime()
          while(true){
              var nowTime = new Date().getTime()
              var ret = ws.read()
              Log("ret:", ret)
              if(nowTime - lastPingTime > pingCyc){
                  var retPing = ws.write("ping")
                  lastPingTime = nowTime
                  Log("Sending: ping", "#FF0000")
              }
              LogStatus("Current time:", _D())
              Sleep(1000)
          }
      }
  }
  
  function onexit() {
      ws.close()
      Log("Exiting")
  }
  

  
      python
      
      
  import json
  import time
  
  ws = None
  def main():
      global ws
      param = {
          "op": "subscribe",
          "args": [{
              "channel": "tickers",
              "instId": "BTC-USDT"
          }]
      }
      ws = Dial("wss://ws.okx.com:8443/ws/v5/public|compress=gzip_raw&mode=recv&reconnect=true&payload=" + json.dumps(param))
      if ws:
          pingCyc = 1000 * 20
          lastPingTime = time.time() * 1000
          while True:
              nowTime = time.time() * 1000
              ret = ws.read()
              Log("ret:", ret)
              if nowTime - lastPingTime > pingCyc:
                  retPing = ws.write("ping")
                  lastPingTime = nowTime
                  Log("Sending: ping", "#FF0000")
              LogStatus("Current time:", _D())
              Sleep(1000)
  
  def onexit():
      ws.close()
      Log("Exiting")
  

  
      c++
      
      
  auto objWS = Dial("wss://ws.okx.com:8443/ws/v5/public|compress=gzip_raw&mode=recv&reconnect=true");
  
  void main() {
      json param = R"({
          "op": "subscribe",
          "args": [{
              "channel": "tickers",
              "instId": "BTC-USDT"
          }]
      })"_json;
  
      objWS.write(param.dump());
      if(objWS.Valid) {
          uint64_t pingCyc = 1000 * 20;
          uint64_t lastPingTime = Unix() * 1000;
          while(true) {
              uint64_t nowTime = Unix() * 1000;
              auto ret = objWS.read();
              Log("ret:", ret);
              if(nowTime - lastPingTime > pingCyc) {
                  auto retPing = objWS.write("ping");
                  lastPingTime = nowTime;
                  Log("Sending: ping", "#FF0000");
              }
              LogStatus("Current time:", _D());
              Sleep(1000);
          }
      }
  }
  
  void onexit() {
      objWS.close();
      Log("Exiting");
  }
  

• Accessing Huobi's WebSocket market data interface:

  • javascript
  • python
  • cpp

  
      javascript
      
      
  var ws = null
  
  function main(){
      var param = {"sub": "market.btcusdt.detail", "id": "id1"}
      ws = Dial("wss://api.huobi.pro/ws|compress=gzip&mode=recv&reconnect=true&payload=" + JSON.stringify(param))
      if(ws){
          while(1){
              var ret = ws.read()
              Log("ret:", ret)
              // Respond to heartbeat packet
              try {
                  var jsonRet = JSON.parse(ret)
                  if(typeof(jsonRet.ping) == "number") {
                      var strPong = JSON.stringify({"pong" : jsonRet.ping})
                      ws.write(strPong)
                      Log("Responding to ping, sending pong:", strPong, "#FF0000")
                  }
              } catch(e) {
                  Log("e.name:", e.name, "e.stack:", e.stack, "e.message:", e.message)
              }
  
              LogStatus("Current time:", _D())
              Sleep(1000)
          }
      }
  }
  
  function onexit() {
      ws.close()
      Log("Executing ws.close()")
  }
  

  
      python
      
      
  import json
  ws = None
  
  def main():
      global ws
      param = {"sub" : "market.btcusdt.detail", "id" : "id1"}
      ws = Dial("wss://api.huobi.pro/ws|compress=gzip&mode=recv&reconnect=true&payload=" + json.dumps(param))
      if ws:
          while True:
              ret = ws.read()
              Log("ret:", ret)
              # Respond to heartbeat packet
              try:
                  jsonRet = json.loads(ret)
                  if "ping" in jsonRet and type(jsonRet["ping"]) == int:
                      strPong = json.dumps({"pong" : jsonRet["ping"]})
                      ws.write(strPong)
                      Log("Responding to ping, sending pong:", strPong, "#FF0000")
              except Exception as e:
                  Log("e:", e)
  
              LogStatus("Current time:", _D())
              Sleep(1000)
  
  def onexit():
      ws.close()
      Log("Executing ws.close()")
  

  
      c++
      
      
  using namespace std;
  
  void main() {
      json param = R"({"sub" : "market.btcusdt.detail", "id" : "id1"})"_json;
      auto ws = Dial("wss://api.huobi.pro/ws|compress=gzip&mode=recv&reconnect=true&payload=" + param.dump());
      if(ws.Valid) {
          while(true) {
              auto ret = ws.read();
              Log("ret:", ret);
              // Respond to heartbeat packet
              try
              {
                  auto jsonRet = json::parse(ret);
                  if(jsonRet["ping"].is_number()) {
                      json pong = R"({"pong" : 0})"_json;
                      pong["pong"] = jsonRet["ping"];
                      auto strPong = pong.dump();
                      ws.write(strPong);
                      Log("Responding to ping, sending pong:", strPong, "#FF0000");
                  }
              } catch(exception &e)
              {
                  Log("e:", e.what());
              }
  
              LogStatus("Current time:", _D());
              Sleep(1000);
          }
      }
  }
  
  void onexit() {
      // ws.close();
      Log("Executing ws.close()");
  }
  

• Accessing OKX's WebSocket authentication interface:

  • javascript
  • python
  • cpp

  
      javascript
      
      
  function getLogin(pAccessKey, pSecretKey, pPassphrase) {
      // Signature function for login
      var ts = (new Date().getTime() / 1000).toString()
      var login = {
          "op": "login",
          "args":[{
              "apiKey"    : pAccessKey,
              "passphrase" : pPassphrase,
              "timestamp" : ts,
              "sign" : exchange.Encode("sha256", "string", "base64", ts + "GET" + "/users/self/verify", "string", pSecretKey)
          }]
      }
      return login
  }
  
  var client_private = null
  function main() {
      // Since the read function uses timeout settings, timeout errors need to be filtered, otherwise redundant error output will be generated
      SetErrorFilter("timeout")
  
      // Position channel subscription information
      var posSubscribe = {
          "op": "subscribe",
          "args": [{
              "channel": "positions",
              "instType": "ANY"
          }]
      }
  
      var accessKey = "xxx"
      var secretKey = "xxx"
      var passphrase = "xxx"
  
      client_private = Dial("wss://ws.okx.com:8443/ws/v5/private")
      client_private.write(JSON.stringify(getLogin(accessKey, secretKey, passphrase)))
      Sleep(3000)  // Cannot subscribe to private channels immediately after login, need to wait for server response
      client_private.write(JSON.stringify(posSubscribe))
      if (client_private) {
          var lastPingTS = new Date().getTime()
          while (true) {
              var buf = client_private.read(-1)
              if (buf) {
                  Log(buf)
              }
  
              // Detect disconnection and reconnect
              if (buf == "" && client_private.write(JSON.stringify(posSubscribe)) == 0) {
                  Log("Detected disconnection, closing connection, reconnecting")
                  client_private.close()
                  client_private = Dial("wss://ws.okx.com:8443/ws/v5/private")
                  client_private.write(JSON.stringify(getLogin(accessKey, secretKey, passphrase)))
                  Sleep(3000)
                  client_private.write(JSON.stringify(posSubscribe))
              }
  
              // Send heartbeat packet
              var nowPingTS = new Date().getTime()
              if (nowPingTS - lastPingTS > 10 * 1000) {
                  client_private.write("ping")
                  lastPingTS = nowPingTS
              }
          }
      }
  }
  
  function onexit() {
      var ret = client_private.close()
      Log("Connection closed!", ret)
  }
  

  
      python
      
      
  import json
  import time
  
  def getLogin(pAccessKey, pSecretKey, pPassphrase):
      ts = str(time.time())
      login = {
          "op": "login",
          "args":[{
              "apiKey"    : pAccessKey,
              "passphrase" : pPassphrase,
              "timestamp" : ts,
              "sign" : exchange.Encode("sha256", "string", "base64", ts + "GET" + "/users/self/verify", "string", pSecretKey)
          }]
      }
      return login
  
  client_private = None
  def main():
      global client_private
      SetErrorFilter("timeout")
  
      posSubscribe = {
          "op": "subscribe",
          "args": [{
              "channel": "positions",
              "instType": "ANY"
          }]
      }
  
      accessKey = "xxx"
      secretKey = "xxx"
      passphrase = "xxx"
  
      client_private = Dial("wss://ws.okx.com:8443/ws/v5/private")
      client_private.write(json.dumps(getLogin(accessKey, secretKey, passphrase)))
      Sleep(3000)
      client_private.write(json.dumps(posSubscribe))
      if client_private:
          lastPingTS = time.time() * 1000
          while True:
              buf = client_private.read(-1)
              if buf:
                  Log(buf)
  
              if buf == "" and client_private.write(json.dumps(posSubscribe)) == 0:
                  Log("Detected disconnection, closing connection, reconnecting")
                  ret = client_private.close()
                  client_private = Dial("wss://ws.okx.com:8443/ws/v5/private")
                  client_private.write(json.dumps(getLogin(accessKey, secretKey, passphrase)))
                  Sleep(3000)
                  client_private.write(json.dumps(posSubscribe))
  
              nowPingTS = time.time() * 1000
              if nowPingTS - lastPingTS > 10 * 1000:
                  client_private.write("ping")
                  lastPingTS = nowPingTS
  
  def onexit():
      ret = client_private.close()
      Log("Connection closed!", ret)
  

  
      c++
      
      
  auto client_private = Dial("wss://ws.okx.com:8443/ws/v5/private");
  
  json getLogin(string pAccessKey, string pSecretKey, string
  pPassphrase) {
      auto ts = std::to_string(Unix());
      json login = R"({
          "op": "login",
          "args": [{
              "apiKey": "",
              "passphrase": "",
              "timestamp": "",
              "sign": ""
          }]
      })"_json;
      login["args"][0]["apiKey"] = pAccessKey;
      login["args"][0]["passphrase"] = pPassphrase;
      login["args"][0]["timestamp"] = ts;
      login["args"][0]["sign"] = exchange.Encode("sha256", "string", "base64", ts + "GET" + "/users/self/verify", "string", pSecretKey);
      return login;
  }
  
  void main() {
      SetErrorFilter("timeout");
      json posSubscribe = R"({
          "op": "subscribe",
          "args": [{
              "channel": "positions",
              "instType": "ANY"
          }]
      })"_json;
  
      auto accessKey = "xxx";
      auto secretKey = "xxx";
      auto passphrase = "xxx";
  
      client_private.write(getLogin(accessKey, secretKey, passphrase).dump());
      Sleep(3000);
      client_private.write(posSubscribe.dump());
  
      if (client_private.Valid) {
          uint64_t lastPingTS = Unix() * 1000;
  
          while (true) {
              auto buf = client_private.read(-1);
              if (buf != "") {
                  Log(buf);
              }
              if (buf == "") {
                  if (client_private.write(posSubscribe.dump()) == 0) {
                      Log("Detected disconnection, closing connection, reconnecting");
                      client_private.close();
                      client_private = Dial("wss://ws.okx.com:8443/ws/v5/private");
                      client_private.write(getLogin(accessKey, secretKey, passphrase).dump());
                      Sleep(3000);
                      client_private.write(posSubscribe.dump());
                  }
              }
  
              uint64_t nowPingTS = Unix() * 1000;
              if (nowPingTS - lastPingTS > 10 * 1000) {
                  client_private.write("ping");
                  lastPingTS = nowPingTS;
              }
          }
      }
  }
  
  void onexit() {
      client_private.close();
      Log("Exiting");
  }
  

• Access CoinEx's WebSocket authentication interface:

  • javascript
  • python
  • cpp

  
      javascript
      
      
  var conn = null
  function main() {
      var accessKey = "your accessKey"
  
      var ts = new Date().getTime()
      var signature = exchange.Encode("sha256", "string", "hex", String(ts), "string", "{{secretkey}}")
      Log("signature:", signature)
  
      var payload = {
          "id": 1,
          "method": "server.sign",
          "params": {
              "access_id": accessKey,
              "signed_str": signature,
              "timestamp": ts,
          }
      }
      Log(`JSON.stringify(payload):`, JSON.stringify(payload))
  
      conn = Dial("wss://socket.coinex.com/v2/futures|compress=gzip&mode=recv&payload=" + JSON.stringify(payload))
      if (!conn) {
          throw "stop"
      }
      Log("Dial ... ", conn.read())
  
      // Subscribe to position updates
      conn.write(JSON.stringify({
          "method": "position.subscribe",
          "params": {"market_list": ["BTCUSDT"]},
          "id": 1
      }))
  
      while (true) {
          var msg = conn.read()
          if (msg) {
              Log("msg:", msg)
          }
      }
  }
  
  function onexit() {
      conn.close()
  }
  

  
      python
      
      
  // Omitted

  
      c++
      
      
  // Omitted

• Example of accessing the MEXC exchange Websocket interface, subscribing to the public.aggre.deals.v3.api.pb channel, and using protobuf.js to decode binary data:

  • javascript
  • python
  • cpp

  
      javascript
      
      
  let strPushDataV3ApiWrapper = `syntax = "proto3";
  
  
  
  option java_package = "com.mxc.push.common.protobuf";
  option optimize_for = SPEED;
  option java_multiple_files = true;
  option java_outer_classname = "PushDataV3ApiWrapperProto";
  
  message PublicAggreDealsV3Api {
  
    repeated PublicAggreDealsV3ApiItem deals  = 1;
    string eventType = 2;
  }
  
  message PublicAggreDealsV3ApiItem {
    string price = 1;
    string quantity = 2;
    int32 tradeType = 3;
    int64 time = 4;
  }
  
  message PushDataV3ApiWrapper {
    string channel = 1;
    oneof body {
      PublicAggreDealsV3Api publicAggreDeals = 314;
    }
  
    optional string symbol = 3;
    optional string symbolId = 4;
    optional int64 createTime = 5;
    optional int64 sendTime = 6;
  }`
  
  let code = HttpQuery("https://cdnjs.cloudflare.com/ajax/libs/protobufjs/7.5.3/protobuf.js")
  let exports = {}
  let module = { exports }
  new Function("module", "exports", code)(module, exports)
  let protobuf = module.exports
  
  function main() {
      const PushDataV3ApiWrapper = protobuf.parse(strPushDataV3ApiWrapper).root.lookupType("PushDataV3ApiWrapper")
  
      var payload = {
          "method": "SUBSCRIPTION",
          "params": [
              "[email protected]@100ms@BTCUSDT"
          ]
      }
  
      // proxy=socks5://x.x.x.x:xxxx
      var conn = Dial("wss://wbs-api.mexc.com/ws|payload=" + JSON.stringify(payload))
  
      var data = null
      while (true) {
          var ret = conn.read()
          if (ret) {
              const uint8arrayData = new Uint8Array(ret)
              const message = PushDataV3ApiWrapper.decode(uint8arrayData)
  
              data = PushDataV3ApiWrapper.toObject(message, {
                longs: String,
                enums: String,
                bytes: String,
                defaults: true,
                arrays: true,
                objects: true
              })
              Log("data:", data)
          }
          LogStatus(_D(), data)
      }
  }
  

  
      python
      
      
  # You can use the corresponding libraries in Python to implement encoding and decoding.

  
      c++
      
      
  // Omitted

• The connection object returned by the Dial function when connecting to a database has 2 unique method functions:

  • exec(sqlString): Used to execute SQL statements, similar to the DBExec() function.

  • fd(): The fd() function returns a handle (e.g., handle variable is handle), which can be used by other threads to reconnect (even if the object created by Dial has already executed the close() function to close the connection). Pass the handle to the Dial() function, e.g., Dial(handle) to reuse the connection.

  The following is an example of using the Dial function to connect to a sqlite3 database.

  • javascript
  • python
  • cpp

  
      javascript
      
      
  var client = null
  function main() {
      // client = Dial("sqlite3://:memory:")   // Use in-memory database
      client = Dial("sqlite3://test1.db")      // Open/connect to the database file in the docker's directory
  
      // Record the handle
      var sqlite3Handle = client.fd()
      Log("sqlite3Handle:", sqlite3Handle)
  
      // Query tables in the database
      var ret = client.exec("SELECT name FROM sqlite_master WHERE type='table'")
      Log(ret)
  }
  
  function onexit() {
      Log("Executing client.close()")
      client.close()
  }
  

  
      python
      
      
  // Not supported

  
      c++
      
      
  // Not supported

Returns

Arguments

Remarks

Detailed explanation of the address parameter: After the normal address wss://ws.okx.com:8443/ws/v5/public, use the | symbol as a separator. If the parameter string contains the | character, use || as the separator. The part after it is the function parameter settings, with each parameter connected by the & character.

For example, when setting both ss5 proxy and compression parameters simultaneously, you can write:
Dial("wss://ws.okx.com:8443/ws/v5/public|proxy=socks5://xxx:9999&compress=gzip_raw&mode=recv")

The Dial() function only supports live trading.

When using the Dial function to connect to a database, please refer to the Go language driver project for each database for writing the connection string.

Note that when the payload content set in the address parameter contains the character = or other special characters, it may affect the parsing of the address parameter in the Dial function. For example:

BackPack Exchange WebSocket private interface call example:

            javascript
            
            
var client = null

function main() {
    // base64-encoded public key of the key pair, i.e., the access key configured on FMZ
    var base64ApiKey = "xxx"

    var ts = String(new Date().getTime())
    var data = "instruction=subscribe&timestamp=" + ts + "&window=5000"

    // Since signEd25519 ultimately returns base64 encoding, it may contain the character "="
    var signature = signEd25519(data)

    // payload may contain the character "=" after JSON encoding
    payload = {
        "method": "SUBSCRIBE",
        "params": ["account.orderUpdate"],
        "signature": [base64ApiKey, signature, ts, "5000"]
    }

    client = Dial("wss://ws.backpack.exchange")
    client.write(JSON.stringify(payload))
    if (!client) {
        Log("Connection failed, exiting")
        return
    }

    while (true) {
        var buf = client.read()
        Log(buf)
    }
}

function onexit() {
    client.close()
}

function signEd25519(data) {
    return exchange.Encode("ed25519.seed", "raw", "base64", data, "base64", "{{secretkey}}")
}

The following calling method in the code works correctly:

            javascript
            
            
client = Dial("wss://ws.backpack.exchange")
client.write(JSON.stringify(payload))

If written directly in payload, it will not work properly, for example:

            javascript
            
            
client = Dial("wss://ws.backpack.exchange|payload=" +
JSON.stringify(payload))

Currently, only JavaScript supports using mqtt, nats, amqp, kafka communication protocols in the Dial function. The following uses JavaScript strategy code as an example to demonstrate usage examples of the four protocols mqtt, nats, amqp, kafka:

            javascript
            
            
// You need to configure and deploy the proxy servers for each protocol first
// For demonstration purposes, the subscription (read operation) and publishing (write operation) of topic test_topic are both performed in this strategy

var arrConn = []
var arrName = []

function main() {
    LogReset(1)
    conn_nats = Dial("nats://[email protected]:4222?topic=test_topic")
    conn_mqtt = Dial("mqtt://127.0.0.1:1883?topic=test_topic")
    conn_amqp = Dial("amqp://q:[email protected]:5672/?queue=test_Queue")
    conn_kafka = Dial("kafka://localhost:9092/test_topic")
    arrConn = [conn_nats, conn_amqp, conn_mqtt, conn_kafka]
    arrName = ["nats", "amqp", "mqtt", "kafka"]

    while (true) {
        for (var i in arrConn) {
            var conn = arrConn[i]
            var name = arrName[i]

            // Write data
            conn.write(name + ", time: " + _D() + ", test msg.")

            // Read data
            var readMsg = conn.read(1000)
            Log(name + " readMsg: ", readMsg, "#FF0000")
        }

        Sleep(1000)
    }
}

function onexit() {
    for (var i in arrConn) {
        arrConn[i].close()
        Log("Closing", arrName[i], "connection")
    }
}

For detailed documentation, please refer to: Exploring FMZ: Communication Protocol Practice Between Live Trading Strategies