fix 接受消息

Signed-off-by: sdvdxl杜龙少 <sdvdxl@163.com>

src/main/java/me/hekr/demo/GatewayDevice.java

@@ -5,6 +5,7 @@ import com.alibaba.fastjson.JSONObject;
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
@@ -12,7 +13,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 public class GatewayDevice {
-  private static final Executor EXECUTOR = Executors.newSingleThreadExecutor();
+  private static final Executor EXECUTOR = Executors.newCachedThreadPool();
   // 设备id
   private static final String DEV_TID = "";
   // 产品密钥
@@ -26,6 +27,7 @@ public class GatewayDevice {
 
   public static void main(String[] args)
       throws IOException, InterruptedException, ExecutionException, TimeoutException {
+    CountDownLatch countDownLatch = new CountDownLatch(1);
     TcpClient tcpClient = new TcpClient("hub.hekr.me", 83).connect();
 
     String randomKey = null;
@@ -39,9 +41,11 @@ public class GatewayDevice {
       message.put("msgId", tcpClient.getMsgId());
       params = new HashMap<>();
       params.put("devTid", DEV_TID);
+      params.put("prodKey", PROD_KEY);
       message.put("params", params);
       tcpClient.send(JSON.toJSONString(message) + "\n");
       String respMsg = tcpClient.receive(MAX_WAIT_TIME_MILLS);
+      System.out.println(respMsg);
       JSONObject respMap = JSON.parseObject(respMsg);
       boolean success = respMap.getIntValue("code") == 200;
       if (!success) {
@@ -50,7 +54,7 @@ public class GatewayDevice {
       }
 
       // 获取 randomKey
-      randomKey = (String) respMap.get("randomKey");
+      randomKey = (String) respMap.getJSONObject("params").get("randomKey");
     }
 
     // 网关登录
@@ -58,6 +62,7 @@ public class GatewayDevice {
     params.put("devTid", DEV_TID);
     // 如果启用两步认证，需要加上authKey
     if (GATEWAY_DEVICE_TWO_FACTOR_AUTHENTICATION) {
+      System.out.println("randomKey:" + randomKey);
       String authKey = Util.MD5(randomKey + DEV_TID + PRIV_KEY);
       params.put("authKey", authKey);
     }
@@ -67,6 +72,7 @@ public class GatewayDevice {
     message.put("params", params);
     tcpClient.send(JSON.toJSONString(message) + "\n");
     String respMsg = tcpClient.receive(MAX_WAIT_TIME_MILLS);
+    System.out.println(respMsg);
     JSONObject resp = JSON.parseObject(respMsg);
     if (resp.getIntValue("code") != 200) {
       System.out.println("gateway login failed,resp:" + respMsg);
@@ -74,12 +80,14 @@ public class GatewayDevice {
     }
 
     // 登录成功，这里订阅消息
-    subcribeMessage(tcpClient);
+    subscribeMessage(tcpClient);
 
     // TODO 发送业务数据
 
     // 循环定时发送心跳
     loopSendHeartbeat(tcpClient);
+
+    countDownLatch.await();
   }
 
   private static void loopSendHeartbeat(TcpClient tcpClient) {
@@ -91,7 +99,7 @@ public class GatewayDevice {
             heartbeatMessage.put("action", "heartbeat");
             heartbeatMessage.put("msgId", tcpClient.getMsgId());
             try {
-              tcpClient.send(JSON.toJSONString(heartbeatMessage));
+              tcpClient.send(JSON.toJSONString(heartbeatMessage) + "\n");
             } catch (IOException e) {
               System.out.println(e.getMessage());
             }
@@ -100,10 +108,12 @@ public class GatewayDevice {
             } catch (InterruptedException ignored) {
             }
           }
+
+          System.out.println("tcp client closed");
         });
   }
 
-  private static void subcribeMessage(TcpClient tcpClient) {
+  private static void subscribeMessage(TcpClient tcpClient) {
     EXECUTOR.execute(
         new ReceiveMessageTask(
             tcpClient,

src/main/java/me/hekr/demo/TcpClient.java

@@ -14,7 +14,6 @@ import java.util.concurrent.TimeoutException;
 
 /** 描述 Created by du on 2017/8/6. */
 public class TcpClient {
-  private static final int TIMEOUT_MILL = 20000;
   private Socket client;
   private String host;
   private int port;
@@ -46,7 +45,7 @@ public class TcpClient {
 
   public TcpClient connect(int timeout) throws IOException {
     client.setKeepAlive(true);
-    client.setSoTimeout(TIMEOUT_MILL);
+    client.setSoTimeout(0);
     client.connect(new InetSocketAddress(host, port), timeout);
 
     outToServer = new DataOutputStream(client.getOutputStream());
@@ -62,6 +61,7 @@ public class TcpClient {
       throw new RuntimeException("output channel closed");
     }
 
+    System.out.println("SEND => " + message);
     outToServer.writeBytes(message);
     return this;
   }
@@ -107,7 +107,8 @@ public class TcpClient {
    */
   public String receive(int timeoutMill)
       throws InterruptedException, ExecutionException, TimeoutException {
-    return new FutureTask<String>(() -> inFromServer.readLine())
-        .get(timeoutMill, TimeUnit.MICROSECONDS);
+    FutureTask<String> futureTask = new FutureTask<>(() -> inFromServer.readLine());
+    futureTask.run();
+    return futureTask.get(timeoutMill, TimeUnit.MILLISECONDS);
   }
 }