Dial

用于原始 Socket 访问，支持 tcp、udp、tls、unix 协议。支持 4 种主流通信协议：mqtt、nats、amqp、kafka。同时支持连接数据库，可用的数据库包括：sqlite3、mysql、postgres、clickhouse。

    
    
    
Dial(address)

    
    
    
Dial(address, timeout)

    
    
    
Dial(address, options)

示例

• Dial 函数调用示例：

  • javascript
  • python
  • rust
  • cpp

  
      javascript
      
      
  function main(){
      // Dial 支持 tcp://、udp://、tls://、unix:// 协议，可传入一个参数指定超时秒数
      var client = Dial("tls://www.baidu.com:443")
      if (client) {
          // write 可额外传入一个数字参数指定超时，返回成功发送的字节数
          client.write("GET / HTTP/1.1\nConnection: Closed\n\n")
          while (true) {
              // read 可额外传入一个数字参数指定超时，单位：毫秒；返回 null 表示出错、超时或 socket 已关闭
              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()
  

  
      rust
      
      
  fn main() {
      // Dial 支持 tcp://、udp://、tls://、unix:// 协议，可使用 Dial::new(addr, timeout) 指定超时秒数
      let mut client = Dial("tls://www.baidu.com:443");
      if client.Valid() {
          // write 的第二个数字参数用于指定超时，返回成功发送的字节数
          client.write("GET / HTTP/1.1\nConnection: Closed\n\n", 0);
          loop {
              // read 的数字参数用于指定超时，单位：毫秒；返回空字符串表示出错、超时或 socket 已关闭
              let buf = client.read(0);
              if 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();
      }
  }
  

• 访问币安（Binance）的 WebSocket 行情接口：

  • javascript
  • python
  • rust
  • cpp

  
      javascript
      
      
  function main() {
      LogStatus("Connecting...")
      // 访问币安的 WebSocket 接口
      var client = Dial("wss://stream.binance.com:9443/ws/!ticker@arr")
      if (!client) {
          Log("Connection failed, exiting")
          return
      }
  
      while (true) {
          // read 仅返回调用 read 之后接收到的数据
          var buf = client.read()
          if (!buf) {
              break
          }
          var table = {
              type: 'table',
              title: '行情图表',
              cols: ['币种', '最高', '最低', '买一', '卖一', '最后成交价', '成交量', '更新时间'],
              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" : "行情图表",
              "cols" : ["币种", "最高", "最低", "买一", "卖一", "最后成交价", "成交量", "更新时间"],
              "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()
  

  
      rust
      
      
  fn main() {
      LogStatus!("Connecting...");
      // 访问币安的 WebSocket 接口
      let mut client = Dial("wss://stream.binance.com:9443/ws/!ticker@arr");
      if !client.Valid() {
          Log!("Connection failed, exiting");
          return;
      }
  
      loop {
          // read 仅返回调用 read 之后接收到的数据
          let buf = client.read(0);
          if buf == "" {
              break;
          }
          let obj = JSONParse(&buf).unwrap();
          // Rust SDK 没有 JSON 序列化功能，此处使用字符串拼接来构造状态栏表格的 JSON 文本
          let mut rows = String::new();
          if let Some(arr) = obj.as_array() {
              for ticker in arr {
                  if !rows.is_empty() {
                      rows += ",";
                  }
                  rows += &format!(r#"["{}","{}","{}","{}","{}","{}","{}","{}"]"#,
                      ticker["s"].as_str().unwrap_or(""), ticker["h"].as_str().unwrap_or(""),
                      ticker["l"].as_str().unwrap_or(""), ticker["b"].as_str().unwrap_or(""),
                      ticker["a"].as_str().unwrap_or(""), ticker["c"].as_str().unwrap_or(""),
                      ticker["q"].as_str().unwrap_or(""), _D(ticker["E"].as_i64().unwrap_or(0)));
              }
          }
          let table = format!(r#"{{"type":"table","title":"行情图表","cols":["币种","最高","最低","买一","卖一","最后成交价","成交量","更新时间"],"rows":[{}]}}"#, rows);
          LogStatus!(format!("`{}`", 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" : "行情图表",
              "cols" : ["币种", "最高", "最低", "买一", "卖一", "最后成交价", "成交量", "更新时间"],
              "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();
  }
  

• 访问币安（Binance）的 WebSocket 接口，并设置 wss 请求头。

  • javascript
  • python
  • rust
  • 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)
  

  
      rust
      
      
  fn main() {
      // Rust 中使用 Dial::with_options()，以 JSON 字符串形式传入 options 设置请求头
      let options = r#"{"headers": {"X-MBX-APIKEY": "your access key"}}"#;
      let random = format!("fmz{}", UnixNano());
      let ts = Unix() * 1000;
      let secretKey = "your secret key";
      let topic = "com_announcement_en";
      let payload = format!("random={}&topic={}&recvWindow=30000&timestamp={}", random, topic, ts);
      let signature = Encode("sha256", "string", "hex", &payload, "string", secretKey);
      let query = format!("?{}&signature={}", payload, signature);
  
      Log!("query:", query);
      let mut conn = Dial::with_options(&format!("wss://api.binance.com/sapi/wss{}", query), options);
  
      for _i in 0..10 {
          let ret = conn.read(0);
          Log!(ret);
      }
  }
  

  
      c++
      
      
  // 暂不支持

• 访问 OKX 的 WebSocket 行情接口：

  • javascript
  • python
  • rust
  • cpp

  
      javascript
      
      
  var ws = null
  function main(){
      var param = {
          "op": "subscribe",
          "args": [{
              "channel": "tickers",
              "instId": "BTC-USDT"
          }]
      }
      // 调用 Dial 函数时，指定 reconnect=true 即可启用重连模式，指定 payload 即为重连时发送的消息。当 WebSocket 连接断开后，将自动重连并自动发送该消息
      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")
  

  
      rust
      
      
  fn main() {
      let param = r#"{"op":"subscribe","args":[{"channel":"tickers","instId":"BTC-USDT"}]}"#;
      // 调用 Dial 函数时，指定 reconnect=true 即可启用重连模式，指定 payload 即为重连时发送的消息。当 WebSocket 连接断开后，将自动重连并自动发送该消息
      let mut ws = Dial(&format!("wss://ws.okx.com:8443/ws/v5/public|compress=gzip_raw&mode=recv&reconnect=true&payload={}", param));
      if ws.Valid() {
          let pingCyc = 1000 * 20;
          let mut lastPingTime = Unix() * 1000;
          loop {
              let nowTime = Unix() * 1000;
              let ret = ws.read(0);
              Log!("ret:", ret);
              if nowTime - lastPingTime > pingCyc {
                  let retPing = ws.write("ping", 0);
                  lastPingTime = nowTime;
                  Log!("Sending: ping", "#FF0000");
              }
              LogStatus!("Current time:", _D(None));
              Sleep(1000);
          }
      }
      // 在 Rust 中，连接对象在离开作用域时会自动关闭，也可以显式调用 ws.close()
  }
  

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

• 访问火币交易所的 WebSocket 行情接口：

  • javascript
  • python
  • rust
  • 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)
              // 响应心跳包操作
              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)
              # 响应心跳包操作
              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()")
  

  
      rust
      
      
  fn main() {
      let param = r#"{"sub":"market.btcusdt.detail","id":"id1"}"#;
      let mut ws = Dial(&format!("wss://api.huobi.pro/ws|compress=gzip&mode=recv&reconnect=true&payload={}", param));
      if ws.Valid() {
          loop {
              let ret = ws.read(0);
              Log!("ret:", ret);
              // 响应心跳包操作，Rust 中使用 JSONParse() 解析，解析失败返回 None
              if let Some(jsonRet) = JSONParse(&ret) {
                  if jsonRet["ping"].is_number() {
                      let strPong = format!(r#"{{"pong":{}}}"#, jsonRet["ping"].as_i64().unwrap_or(0));
                      ws.write(&strPong, 0);
                      Log!("Responding to ping, sending pong:", strPong, "#FF0000");
                  }
              }
  
              LogStatus!("Current time:", _D(None));
              Sleep(1000);
          }
      }
      // Rust 中连接对象在离开作用域时自动关闭，也可以显式调用 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);
              // 响应心跳包操作
              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()");
  }
  

• 访问 OKX 的 WebSocket 验证接口：

  • javascript
  • python
  • rust
  • cpp

  
      javascript
      
      
  function getLogin(pAccessKey, pSecretKey, pPassphrase) {
      // 签名函数，用于生成登录请求
      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() {
      // 由于 read 函数设置了超时，需过滤超时报错，否则会产生冗余的错误输出
      SetErrorFilter("timeout")
  
      // 持仓频道的订阅信息
      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)  // 登录后不能立即订阅私有频道，需等待服务器响应
      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)
              }
  
              // 检测到连接断开后重连
              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))
              }
  
              // 发送心跳包
              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)
  

  
      rust
      
      
  fn getLogin(pAccessKey: &str, pSecretKey: &str, pPassphrase: &str) -> String {
      // 签名函数，用于生成登录请求。Rust 中没有 exchange.Encode 成员函数，因此使用全局 Encode 函数计算签名
      let ts = format!("{}", Unix());
      let sign = Encode("sha256", "string", "base64", &format!("{}GET/users/self/verify", ts), "string", pSecretKey);
      format!(r#"{{"op":"login","args":[{{"apiKey":"{}","passphrase":"{}","timestamp":"{}","sign":"{}"}}]}}"#, pAccessKey, pPassphrase, ts, sign)
  }
  
  fn main() {
      // 由于 read 函数设置了超时，需过滤超时报错，否则会产生冗余的错误输出
      SetErrorFilter("timeout");
  
      // 持仓频道的订阅信息
      let posSubscribe = r#"{"op":"subscribe","args":[{"channel":"positions","instType":"ANY"}]}"#;
  
      let accessKey = "xxx";
      let secretKey = "xxx";
      let passphrase = "xxx";
  
      let mut client_private = Dial("wss://ws.okx.com:8443/ws/v5/private");
      client_private.write(&getLogin(accessKey, secretKey, passphrase), 0);
      Sleep(3000);  // 登录后不能立即订阅私有频道，需等待服务器响应
      client_private.write(posSubscribe, 0);
      if client_private.Valid() {
          let mut lastPingTS = Unix() * 1000;
          loop {
              let buf = client_private.read(-1);
              if buf != "" {
                  Log!(buf);
              }
  
              // 检测到连接断开后重连
              if buf == "" && client_private.write(posSubscribe, 0) == 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), 0);
                  Sleep(3000);
                  client_private.write(posSubscribe, 0);
              }
  
              // 发送心跳包
              let nowPingTS = Unix() * 1000;
              if nowPingTS - lastPingTS > 10 * 1000 {
                  client_private.write("ping", 0);
                  lastPingTS = nowPingTS;
              }
          }
      }
  }
  

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

• 访问 CoinEx 的 WebSocket 验证接口：

  • javascript
  • python
  • rust
  • 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())
  
      // 订阅持仓推送
      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
      
      
  // 略

  
      rust
      
      
  fn main() {
      let accessKey = "your accessKey";
  
      let ts = Unix() * 1000;
      // Rust 没有 exchange.Encode 成员函数，无法使用 {{secretkey}} 模板替换，使用全局 Encode 函数直接传入秘钥计算签名
      let signature = Encode("sha256", "string", "hex", &format!("{}", ts), "string", "your secretKey");
      Log!("signature:", signature);
  
      // Rust SDK 没有 JSON 序列化功能，使用字符串拼接构造 payload 的 JSON 文本
      let payload = format!(r#"{{"id":1,"method":"server.sign","params":{{"access_id":"{}","signed_str":"{}","timestamp":{}}}}}"#, accessKey, signature, ts);
      Log!("payload:", payload);
  
      let mut conn = Dial(&format!("wss://socket.coinex.com/v2/futures|compress=gzip&mode=recv&payload={}", payload));
      if !conn.Valid() {
          Panic!("stop");
      }
      Log!("Dial ... ", conn.read(0));
  
      // 订阅持仓推送
      conn.write(r#"{"method":"position.subscribe","params":{"market_list":["BTCUSDT"]},"id":1}"#, 0);
  
      loop {
          let msg = conn.read(0);
          if msg != "" {
              Log!("msg:", msg);
          }
      }
  }
  

  
      c++
      
      
  // 略

• 以下示例演示如何访问 MEXC 交易所的Websocket接口，订阅public.aggre.deals.v3.api.pb频道，并使用protobuf.js解码二进制数据：

  • 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
      
      
  # 可以使用 Python 中相应的库实现编码与解码。

  
      c++
      
      
  // 略

• Dial函数连接数据库时返回的连接对象具有2个独有的方法函数：

  • exec(sqlString)：用于执行SQL语句，用法与DBExec()函数类似。

  • fd()：该函数返回一个句柄（例如句柄变量为handle），用于在其它线程中重连。即使由Dial创建的连接对象已通过close()函数关闭，也可将该句柄传入Dial()函数（例如Dial(handle)）以重用连接。

  以下是使用Dial函数连接sqlite3数据库的示例。

  • javascript
  • python
  • rust
  • cpp

  
      javascript
      
      
  var client = null
  function main() {
      // client = Dial("sqlite3://:memory:")   // 使用内存数据库
      client = Dial("sqlite3://test1.db")      // 打开/连接托管者所在目录的数据库文件
  
      // 记录句柄
      var sqlite3Handle = client.fd()
      Log("sqlite3Handle:", sqlite3Handle)
  
      // 查询数据库中的表
      var ret = client.exec("SELECT name FROM sqlite_master WHERE type='table'")
      Log(ret)
  }
  
  function onexit() {
      Log("Executing client.close()")
      client.close()
  }
  

  
      python
      
      
  // 不支持

  
      rust
      
      
  fn main() {
      // let mut client = Dial("sqlite3://:memory:");   // 使用内存数据库
      let mut client = Dial("sqlite3://test1.db");      // 打开/连接托管者所在目录的数据库文件
  
      // Rust 的连接对象不支持 fd() 方法
  
      // 查询数据库中的表
      let ret = client.exec("SELECT name FROM sqlite_master WHERE type='table'");
      Log!(format!("{:?}", ret));
  
      Log!("Executing client.close()");
      client.close();
  }
  

  
      c++
      
      
  // 不支持

返回值

参数

备注

address参数的详细说明：在标准地址wss://ws.okx.com:8443/ws/v5/public之后，使用|符号进行分隔。如果参数字符串中包含|字符，则使用||作为分隔符。分隔符之后的部分为功能参数设置，各参数之间使用&字符连接。

例如，同时设置ss5代理和压缩参数时，可以写作：

Dial("wss://ws.okx.com:8443/ws/v5/public|proxy=socks5://xxx:9999&compress=gzip_raw&mode=recv")

Dial()函数仅支持实盘。

使用Dial函数连接数据库时，连接字符串的编写方式可参考各数据库对应的Go语言驱动项目。

需要注意，当address参数中设置的payload内容包含字符=或其它特殊字符时，可能会影响Dial函数对address参数的解析，示例如下。

backPack交易所websocket私有接口调用示例：

            javascript
            
            
var client = null

function main() {
    // base64编码的秘钥对公钥，即在FMZ上配置的access key
    var base64ApiKey = "xxx"

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

    // 由于signEd25519最终返回的是base64编码，其中会有字符"="
    var signature = signEd25519(data)

    // payload 被JSON编码后可能包含字符"="
    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}}")
}

代码中采用以下调用方式可以正常工作：

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

如果直接写在payload中则无法正常工作，例如：

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

目前仅JavaScript语言支持在Dial函数中使用mqtt、nats、amqp、kafka通信协议，下面以JavaScript语言策略代码为例，演示mqtt、nats、amqp、kafka四种协议的使用方法：

            javascript
            
            
// 需要先配置并部署完成各个协议的代理服务器
// 为了便于演示，主题test_topic的订阅（read操作）与发布（write操作）都在当前这个策略中进行

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]

            // 写数据
            conn.write(name + ", time: " + _D() + ", test msg.")

            // 读数据
            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")
    }
}

详细介绍请参考文档：探索FMZ：交易策略实盘间通信协议实践