feat: 重构客户端

1，完善client端的配置信息； 2，设计将同步数据从内存转入数据库的设计； 3，为客户端增加一个同步线程自动同步数据；

feat: 重构客户端
1，完善client端的配置信息； 2，设计将同步数据从内存转入数据库的设计； 3，为客户端增加一个同步线程自动同步数据；
wu
1 parent 69bfaf7f
Showing 8 changed files with 111 additions and 77 deletions Show diff stats
src/main/java/com/example/mina/client/base/AbstractClient.java
src/main/java/com/example/mina/client/base/AbstractVirtualBoxClientHandler.java
src/main/java/com/example/mina/client/base/BoxConnectConfig.java
src/main/java/com/example/mina/client/base/DownloadListener.java
src/main/java/com/example/mina/client/box/aeroflex/AeroflexClient.java
src/main/java/com/example/mina/client/box/aeroflex/AeroflexClientHandler.java
src/main/java/com/example/mina/client/box/lte3000/Lte3000Client.java
src/main/java/com/example/mina/client/box/lte3000/Lte3000ClientHandler.java
 package com.example.mina.client.base;
-import com.example.mina.server.base.AbstractHardwareDataBuffer;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder;
 import org.apache.mina.core.future.ConnectFuture;
 import org.apache.mina.core.service.IoConnector;
 import org.apache.mina.core.session.IoSession;
-import org.apache.mina.filter.codec.ProtocolCodecFactory;
 import org.apache.mina.filter.codec.ProtocolCodecFilter;
 import org.apache.mina.transport.socket.nio.NioSocketConnector;
 import java.net.InetSocketAddress;
-import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
 /**
  * @author dy
  * @date 2021/3/10
  */
+@Slf4j
 public abstract class AbstractClient {
     /**
@@ -23,58 +25,62 @@ public abstract class AbstractClient {
     protected BoxConnectConfig boxConnectConfig;
     /**
-     * 存储设备返回的响应数据缓存
-     */
-    protected AbstractHardwareDataBuffer responsePool;
-
-    /**
      * 存储客户端的请求指令的队列
      */
-    protected ArrayBlockingQueue<BaseRequest> requestsPool;
+    protected LinkedBlockingQueue<BaseRequest> requestsPool;
     protected IoSession ioSession;
+    private CommandSender sender;
     public AbstractClient(BoxConnectConfig connectConfig) {
         this.boxConnectConfig = connectConfig;
-        this.requestsPool = new ArrayBlockingQueue<BaseRequest>(100);
-        //TODO 初始化响应缓存
-        this.responsePool = null;
+        this.requestsPool = new LinkedBlockingQueue<>();
+        this.sender = new CommandSender(this, this.requestsPool);
     }
     /**
      *获取设备的类型
      */
-    abstract public String getDeviceType();
+    public String getDeviceType() {
+        return boxConnectConfig.getDeviceType();
+    }
     /**
-     * 返回客户端对响应的Handler
-     * @return
+     * 获取设备ID
      */
-    abstract protected AbstractVirtualBoxClientHandler getIoHandler();
+    public String getDeviceId() {
+        return boxConnectConfig.getDeviceId();
+    }
     /**
-     * 返回该盒子的编解码器
+     * 获取设备名称
+     */
+    public String getDeviceName() {
+        return boxConnectConfig.getDeviceName();
+    }
+
+    /**
+     * 返回客户端对响应的Handler
      * @return
      */
-    abstract protected ProtocolCodecFactory getCodecFactory();
+    abstract protected AbstractVirtualBoxClientHandler getIoHandler();
     /**
-     * 返回缓存数据，提供给Request执行；
+     * 返回该盒子的编解码器,默认为tcp的协议，如果是其它协议，可以在自类中重写改方法
      * @return
      */
-    public AbstractHardwareDataBuffer getResponsePool() {
-        return this.responsePool;
+    protected void setFilterChain(DefaultIoFilterChainBuilder filterChainBuilder) {
+        filterChainBuilder.addLast("codec", new ProtocolCodecFilter(new ByteProtocolFactory()));
     }
-
     public void start() {
         //1、创建客户端IoService
         IoConnector connector = new NioSocketConnector();
         //客户端链接超时时间
         connector.setConnectTimeoutMillis(30000);
         //2、客户端过滤器
-        connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(getCodecFactory()));
+        setFilterChain(connector.getFilterChain());
         //3、客户端IoHandler，发生消息
         connector.setHandler(getIoHandler());
@@ -86,7 +92,8 @@ public abstract class AbstractClient {
         // 获取连接会话
         this.ioSession = connectFuture.getSession();
         this.ioSession.setAttribute(BoxConstants.SESSION_CONFIG_NAME, boxConnectConfig);
-
+        //将数据发送线程启动
+        this.sender.start();
     }
     /**
@@ -100,4 +107,29 @@ public abstract class AbstractClient {
         return request.execute();
     }
+    /**
+     * 客户端启动的时候自动为每个客户端启动一个发送线程，将缓存的指令发送出去
+     */
+    private static class CommandSender extends Thread {
+        private AbstractClient client;
+        private LinkedBlockingQueue<BaseRequest> commands;
+
+        public CommandSender(AbstractClient client, LinkedBlockingQueue commands ) {
+            this.client = client;
+            this.commands = commands;
+        }
+
+        @Override
+        public void run() {
+            while (true) {
+                try {
+                    BaseRequest command = commands.poll(5, TimeUnit.SECONDS);
+
+                    client.ioSession.write(command);
+                }catch (Exception e) {
+                    log.error("error occurred when send the request to device, device: " + client.boxConnectConfig, e);
+                }
+            }
+        }
+    }
 }
 package com.example.mina.client.base;
-import com.example.mina.server.base.AbstractHardwareDataBuffer;
 import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.mina.core.buffer.IoBuffer;
@@ -16,14 +15,10 @@ import org.apache.mina.core.session.IoSession;
 @Slf4j
 public abstract class AbstractVirtualBoxClientHandler extends IoHandlerAdapter {
-    protected String deviceType;
+    protected DownloadListener downloadListener;
-    protected AbstractHardwareDataBuffer responsePool;
-
-
-    public AbstractVirtualBoxClientHandler(String deviceType, AbstractHardwareDataBuffer abstractHardwareDataBuffer) {
-        this.deviceType = deviceType;
-        this.responsePool = abstractHardwareDataBuffer;
+    public AbstractVirtualBoxClientHandler(DownloadListener downloadListener) {
+        this.downloadListener = downloadListener;
     }
     protected BoxConnectConfig getConnectConfig(IoSession session) {
@@ -48,7 +43,7 @@ public abstract class AbstractVirtualBoxClientHandler extends IoHandlerAdapter {
         //TODO 客户端将数据服务器的字符串数组读取出来
         byte[] response = null;
-        log.info("the client recieved the device response, deviceType: {}, device:{}, response is: {}", deviceType,
+        log.info("the client recieved the device response, device:{}, response is: {}",
                 getConnectConfig(session), response);
         handleCommandResponse(response);
     }
@@ -72,12 +67,12 @@ public abstract class AbstractVirtualBoxClientHandler extends IoHandlerAdapter {
     @Override
     public void sessionIdle(IoSession session, IdleStatus status) {
-        log.info("the session with box was idle, deviceType:{}, device is {}", deviceType, getConnectConfig(session));
+        log.info("the session with box was idle, device is {}", getConnectConfig(session));
     }
     @Override
     public void sessionOpened(IoSession session) {
-        log.info("the system has connected to device, deviceType: {}, device is {}", deviceType, getConnectConfig(session));
+        log.info("the system has connected to device, device is {}", getConnectConfig(session));
     }
     abstract public boolean handleCommandResponse(byte[] response);
@@ -10,6 +10,12 @@ import java.util.Map;
 @Builder
 public class BoxConnectConfig {
+    private String deviceId;
+
+    private String deviceType;
+
+    private String deviceName;
+
     private String host;
     private Integer port;
@@ -0,0 +1,17 @@
+package com.example.mina.client.base;
+
+public class DownloadListener {
+
+    /**
+     * 当从设备端将数据同步到客户端的时候，将数据直接存入数据中，
+     * 该类会在ioHandler及其子类中被调用
+     * 设置attenuation的值,并将其存储到matrix_data的数据表中
+     */
+    public void setAttenuation(String deviceId, int row, int col, int val) {
+        //TODO 实现将数据写入到数据库的逻辑
+    }
+
+    private String getCurrentThreadId() {
+        return Thread.currentThread().getName();
+    }
+}
@@ -3,8 +3,6 @@ package com.example.mina.client.box.aeroflex;
 import com.example.mina.client.base.AbstractClient;
 import com.example.mina.client.base.AbstractVirtualBoxClientHandler;
 import com.example.mina.client.base.BoxConnectConfig;
-import com.example.mina.client.base.BoxConstants;
-import org.apache.mina.filter.codec.ProtocolCodecFactory;
 /**
  * @author dy
@@ -17,16 +15,6 @@ public class AeroflexClient extends AbstractClient {
     }
     @Override
-    public String getDeviceType() {
-        return BoxConstants.DEVICE_TYPE_AERO;
-    }
-
-    @Override
-    protected ProtocolCodecFactory getCodecFactory() {
-        return null;
-    }
-
-    @Override
     protected AbstractVirtualBoxClientHandler getIoHandler() {
         return null;
     }
 package com.example.mina.client.box.aeroflex;
 import com.example.mina.client.base.AbstractVirtualBoxClientHandler;
-import com.example.mina.client.base.BaseResponse;
+import com.example.mina.client.base.DownloadListener;
 /**
  * @author dy
@@ -9,6 +9,10 @@ import com.example.mina.client.base.BaseResponse;
  */
 public class AeroflexClientHandler extends AbstractVirtualBoxClientHandler {
+    public AeroflexClientHandler(DownloadListener downloadListener) {
+        super(downloadListener);
+    }
+
     @Override
     public boolean handleCommandResponse(byte[] response) {
@@ -3,7 +3,6 @@ package com.example.mina.client.box.lte3000;
 import com.example.mina.client.base.AbstractClient;
 import com.example.mina.client.base.AbstractVirtualBoxClientHandler;
 import com.example.mina.client.base.BoxConnectConfig;
-import org.apache.mina.filter.codec.ProtocolCodecFactory;
 /**
  * @author dy
@@ -15,17 +14,8 @@ public class Lte3000Client extends AbstractClient {
     }
     @Override
-    public String getDeviceType() {
-        return null;
-    }
-
-    @Override
     protected AbstractVirtualBoxClientHandler getIoHandler() {
         return null;
     }
-    @Override
-    protected ProtocolCodecFactory getCodecFactory() {
-        return null;
-    }
 }
	@@ -10,6 +10,12 @@ import java.util.Map;		@@ -10,6 +10,12 @@ import java.util.Map;
10	@Builder	10	@Builder
11	public class BoxConnectConfig {	11	public class BoxConnectConfig {
12		12
		13	+ private String deviceId;
		14	+
		15	+ private String deviceType;
		16	+
		17	+ private String deviceName;
		18	+
13	private String host;	19	private String host;
14		20
15	private Integer port;	21	private Integer port;
@@ -0,0 +1,17 @@		@@ -0,0 +1,17 @@
	1	+package com.example.mina.client.base;
	2	+
	3	+public class DownloadListener {
	4	+
	5	+ /**
	6	+ * 当从设备端将数据同步到客户端的时候，将数据直接存入数据中，
	7	+ * 该类会在ioHandler及其子类中被调用
	8	+ * 设置attenuation的值,并将其存储到matrix_data的数据表中
	9	+ */
	10	+ public void setAttenuation(String deviceId, int row, int col, int val) {
	11	+ //TODO 实现将数据写入到数据库的逻辑
	12	+ }
	13	+
	14	+ private String getCurrentThreadId() {
	15	+ return Thread.currentThread().getName();
	16	+ }
	17	+}
	@@ -3,8 +3,6 @@ package com.example.mina.client.box.aeroflex;		@@ -3,8 +3,6 @@ package com.example.mina.client.box.aeroflex;
3	import com.example.mina.client.base.AbstractClient;	3	import com.example.mina.client.base.AbstractClient;
4	import com.example.mina.client.base.AbstractVirtualBoxClientHandler;	4	import com.example.mina.client.base.AbstractVirtualBoxClientHandler;
5	import com.example.mina.client.base.BoxConnectConfig;	5	import com.example.mina.client.base.BoxConnectConfig;
6	-import com.example.mina.client.base.BoxConstants;
7	-import org.apache.mina.filter.codec.ProtocolCodecFactory;
8		6
9	/**	7	/**
10	* @author dy	8	* @author dy
	@@ -17,16 +15,6 @@ public class AeroflexClient extends AbstractClient {		@@ -17,16 +15,6 @@ public class AeroflexClient extends AbstractClient {
17	}	15	}
18		16
19	@Override	17	@Override
20	- public String getDeviceType() {
21	- return BoxConstants.DEVICE_TYPE_AERO;
22	- }
23	-
24	- @Override
25	- protected ProtocolCodecFactory getCodecFactory() {
26	- return null;
27	- }
28	-
29	- @Override
30	protected AbstractVirtualBoxClientHandler getIoHandler() {	18	protected AbstractVirtualBoxClientHandler getIoHandler() {
31	return null;	19	return null;
32	}	20	}