From 9c58ad1bbcff068694ea2fcd7fea468bf445a4d3 Mon Sep 17 00:00:00 2001
From: liurunyu <lry9898@163.com>
Date: 星期四, 13 二月 2025 16:04:08 +0800
Subject: [PATCH] 通信中间件，启用消息推送测试代码
---
pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/channel/tcp/TcpUnit.java | 62 +++++++++++++++++++++++--------
1 files changed, 46 insertions(+), 16 deletions(-)
diff --git a/pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/channel/tcp/TcpUnit.java b/pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/channel/tcp/TcpUnit.java
index e0d83d6..7528501 100644
--- a/pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/channel/tcp/TcpUnit.java
+++ b/pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/channel/tcp/TcpUnit.java
@@ -13,7 +13,7 @@
import com.dy.common.mw.UnitAdapterInterface;
import com.dy.common.mw.UnitInterface;
-import com.dy.common.mw.UnitStartedCallbackInterface;
+import com.dy.common.mw.UnitCallbackInterface;
@SuppressWarnings("unused")
public class TcpUnit implements UnitInterface {
@@ -22,22 +22,29 @@
private static boolean started = false ;
private TcpUnitAdapter adapter ;
+ private TcpIoHandler tcpIoHandler ;
+ private DataCodecFactory dataCodecFactory ;
private TcpUnit(){} ;
public static TcpUnit getInstance(){
return instance ;
}
-
+
/**
- * 鎶奍oSession浼氳瘽鐨処D灞炴�ц缃埌IoSession灞炴�т腑
+ * 鎶奍oSession浼氳瘽鐨処D灞炴�у強鍗忚鍚嶇О鐗堟湰鍙疯缃埌IoSession灞炴�т腑
+ * @param session
+ * @param rtuAddr
+ * @param protocolName
+ * @param protocolVersion
* @throws Exception
*/
- public void setIoSessionArrId(IoSession session, String id) throws Exception {
- session.setAttribute(TcpIoSessionAttrIdIsRtuAddr.sessionArrIdKey, id) ;
+ public void setIoSessionArrs(IoSession session, String rtuAddr, String protocolName, Short protocolVersion) throws Exception {
+ session.setAttribute(TcpIoSessionAttrIdIsRtuAddr.sessionArrRtuAddr, rtuAddr) ;
+ session.setAttribute(TcpIoSessionAttrIdIsRtuAddr.sessionArrProtocolName, protocolName) ;
+ session.setAttribute(TcpIoSessionAttrIdIsRtuAddr.sessionArrProtocolVersion, protocolVersion) ;
}
-
@Override
public void setAdapter(UnitAdapterInterface adapter) throws Exception {
if(adapter == null){
@@ -61,7 +68,7 @@
/**
* 鍚姩妯″潡
*/
- public void start(UnitStartedCallbackInterface callback) throws Exception {
+ public void start(UnitCallbackInterface callback) throws Exception {
if(!started){
started = true ;
/**
@@ -93,20 +100,25 @@
//寰楀埌缃戠粶 閫氫俊鏁版嵁杩囨护鍣ㄩ摼
DefaultIoFilterChainBuilder chain = acceptor.getFilterChain() ;
- //缂栬В鐮佽繃婊ゅ櫒
- chain.addLast("protocol", new ProtocolCodecFilter(new DataCodecFactory(this.adapter)));
+ //鐢熸垚缂栬В鐮佽繃婊ゅ櫒宸ュ巶绫�
+ dataCodecFactory = new DataCodecFactory(this.adapter) ;
+ //璁剧疆鈥減rotocol鈥濓紝鍔犲叆缂栬В鐮佽繃婊ゅ櫒锛岃繃婊ゅ櫒鍦↖oProcessor绾跨▼涓墽琛�
+ chain.addLast("protocol", new ProtocolCodecFilter(dataCodecFactory));
/*
- * 涓�鑸珽xecutorFilter 閮借鏀惧湪ProtocolCodecFilter 杩囨护鍣ㄧ殑鍚庨潰锛�
- * 涔熷氨鏄笉瑕佽缂栬В鐮佽繍琛屽湪鐙珛鐨勭嚎绋嬩笂锛岃�屾槸瑕佽繍琛屽湪IoProcessor 鎵�鍦ㄧ殑绾跨▼锛�
- * 鍥犱负缂栬В鐮佸鐞嗙殑鏁版嵁閮芥槸鐢盜oProcessor 璇诲彇鍜屽彂閫佺殑锛屾病蹇呰寮�鍚柊鐨勭嚎绋嬶紝
- * 鍚﹀垯鎬ц兘鍙嶈�屼細涓嬮檷銆備竴鑸娇鐢‥xecutorFilter 鐨勫吀鍨嬪満鏅槸灏嗕笟鍔￠�昏緫锛堣濡傦細鑰楁椂鐨勬暟鎹簱鎿嶄綔锛�
- * 鏀惧湪鍗曠嫭鐨勭嚎绋嬩腑杩愯锛屼篃灏辨槸璇翠笌IO 澶勭悊鏃犲叧鐨勬搷浣滃彲浠ヨ�冭檻浣跨敤ExecutorFilter 鏉ュ紓姝ユ墽琛屻��
+ * 涓�鑸珽xecutorFilter 閮借鏀惧湪ProtocolCodecFilter杩囨护鍣ㄧ殑鍚庨潰锛�
+ * 涔熷氨鏄缂栬В鐮佽繍琛屽湪IoProcessor鎵�鍦ㄧ殑绾跨▼锛屽洜涓虹紪瑙ｇ爜澶勭悊鐨勬暟鎹兘鏄�
+ * 鐢盜oProcessor璇诲彇鍜屽彂閫佺殑锛屾病蹇呰寮�鍚柊鐨勭嚎绋嬶紝鍚﹀垯鎬ц兘鍙嶈�屼細涓嬮檷銆�
+ * ExecutorFilter杩囩▼鍣ㄤ細鍚姩涓�涓嚎绋嬫睜锛屽鐞嗗悗缁唬鐮侀�昏緫銆�
+ * 涓�鑸娇鐢‥xecutorFilter鐨勫吀鍨嬪満鏅槸灏嗕笟鍔￠�昏緫锛堣濡傦細鑰楁椂鐨勬暟鎹簱鎿嶄綔锛�
+ * 鏀惧湪鍗曠嫭鐨勭嚎绋嬩腑杩愯锛屼篃灏辨槸璇翠笌IO澶勭悊鏃犲叧鐨勬搷浣滃彲浠ヨ�冭檻浣跨敤ExecutorFilter鏉ュ紓姝ユ墽琛屻��
+ * 鏈鐢ㄦ硶锛屼娇ExecutorFilter绾跨▼姹犱腑鐨勭嚎绋嬪鐞咺OHandler锛圱cpIoHandler锛夋搷浣�
*/
chain.addLast("exceutor", new ExecutorFilter());
//涓氬姟閫昏緫澶勭悊鍣紝璐熻矗澶勭悊缃戠粶浼氳瘽鍙婅緭鍏ヨ緭鍑烘暟鎹�
- acceptor.setHandler(new TcpIoHandler(this.adapter));
+ tcpIoHandler = new TcpIoHandler(this.adapter) ;
+ acceptor.setHandler(tcpIoHandler) ;
boolean isException = false ;
try {
@@ -129,8 +141,26 @@
}
}
+ /**
+ * 鍋滄妯″潡杩愯锛屽皢涓嶅啀鎺ュ叆TCP缃戠粶杩炴帴锛屽苟鎶婂凡缁弔cp杩炴帴鐨勫叏閮ㄦ柇杩炴帴
+ * @param callback
+ * @throws Exception
+ */
@Override
- public void stop(UnitStartedCallbackInterface callback) throws Exception {
+ public void stop(UnitCallbackInterface callback) throws Exception {
+ this.tcpIoHandler.stop();
+ this.dataCodecFactory.stop();
+ this.adapter.newUnitStopCallback().callback();
+ callback.call(null);
+ }
+
+ /**
+ * 瑙ｉ櫎鍋滄锛屾仮澶峊CP鏈嶅姟杩愯
+ * @throws Exception
+ */
+ public void recover() throws Exception {
+ this.tcpIoHandler.recover();
+ this.dataCodecFactory.recover();
}
--
Gitblit v1.8.0