close channel group

Author: caoliang1918

README.md

@@ -13,7 +13,7 @@ With `Maven Central` repositories (stable releases only):
     <dependency>
         <groupId>org.zhongweixian</groupId>
         <artifactId>websocket-api</artifactId>
-        <version>2.0.2</version>
+        <version>2.0.3</version>
     </dependency>
 ```
 

pom.xml

@@ -5,7 +5,7 @@
 
     <groupId>org.zhongweixian</groupId>
     <artifactId>websocket-api</artifactId>
-    <version>2.0.2</version>
+    <version>2.0.3</version>
     <packaging>jar</packaging>
 
     <name>websocket-client</name>
@@ -91,7 +91,7 @@
                     <plugin>
                         <groupId>org.apache.maven.plugins</groupId>
                         <artifactId>maven-source-plugin</artifactId>
-                        <version>3.0.1</version>
+                        <version>3.1.0</version>
                         <executions>
                             <execution>
                                 <id>attach-sources</id>

src/main/java/org/zhongweixian/client/tcp/NettyClient.java

@@ -137,7 +137,7 @@ private void connect() {
             channel = channelFuture.channel();
             channelFuture.channel().closeFuture().sync();
         } catch (InterruptedException e) {
-            logger.error("{}", e);
+            logger.error(e.getMessage(), e);
         } finally {
             if (channelFuture != null) {
                 if (channelFuture.channel() != null && channelFuture.channel().isOpen()) {
@@ -154,7 +154,7 @@ private void connect() {
                 }
                 TimeUnit.SECONDS.sleep(sleep);
             } catch (InterruptedException e) {
-                logger.error("{}", e);
+                logger.error(e.getMessage(), e);
             }
             if (!autoReConnect) {
                 return;
@@ -199,6 +199,7 @@ public void operationComplete(ChannelFuture future) throws Exception {
     public void close() {
         if (channel != null && channel.isOpen()) {
             channel.close();
+            group.shutdownGracefully();
         }
         autoReConnect = false;
         logger.info("client close {}:{} , autoReConnect:{}", host, port, autoReConnect);

src/main/java/org/zhongweixian/client/tcp/handler/SimpleClientHandler.java

@@ -83,7 +83,7 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception {
         if (authorizationToken.getTimeHeart()) {
             logger.info("start send ping, timeHeart:{}s, channel:{}", authorizationToken.getHeart(), ctx.channel().remoteAddress());
             if (executorService != null) {
-                executorService.isShutdown();
+                executorService.shutdownNow();
             }
             executorService = Executors.newScheduledThreadPool(1);
             executorService.scheduleAtFixedRate(() -> {
@@ -106,20 +106,24 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception {
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
         //异常时断开连接
-        logger.error("exceptionCaught:{}", cause);
+        logger.error("exceptionCaught", cause);
+        if (executorService != null) {
+            logger.info("shutdown thread:{}", ctx.channel());
+            executorService.shutdownNow();
+        }
         active = false;
         listener.onClose(ctx.channel(), 501, cause.getMessage());
         ctx.close();
     }
 
     @Override
     public void channelInactive(ChannelHandlerContext ctx) {
-        logger.error("socket close, channel active : {}", ctx.channel().isActive());
+        logger.error("socket close, channel active:{}", ctx.channel().isActive());
         active = false;
-        listener.onClose(ctx.channel(), 500, "connect to server close");
         if (executorService != null) {
             executorService.shutdownNow();
         }
+        listener.onClose(ctx.channel(), 500, "connect to server close");
         ctx.channel().close();
     }
 }

src/main/java/org/zhongweixian/client/websocket/WebSocketClient.java

@@ -38,6 +38,7 @@ public class WebSocketClient {
     private SslContext sslContext;
     private EventLoopGroup group = null;
     private Bootstrap bootstrap = null;
+    private Channel channel;
 
     private static final int HEART_TIME = 10;
 
@@ -105,12 +106,14 @@ public void operationComplete(ChannelFuture channelFuture) throws Exception {
                 }
             }
         });
+        this.channel = channel;
         return new WsConnection(channel);
     }
 
 
     public void close() {
-        if (group != null) {
+        if (channel != null && channel.isOpen()) {
+            channel.close();
             group.shutdownGracefully();
         }
     }

src/main/java/org/zhongweixian/client/websocket/WsClient.java

@@ -115,7 +115,7 @@ private void connect() {
             }
             channelFuture.channel().closeFuture().sync();
         } catch (Exception e) {
-            logger.error("{}", e);
+            logger.error(e.getMessage(), e);
         } finally {
             if (channel != null && channel.isOpen()) {
                 channel.close();
@@ -127,7 +127,7 @@ private void connect() {
             try {
                 TimeUnit.SECONDS.sleep(heart);
             } catch (InterruptedException e) {
-                logger.error("{}", e);
+                logger.error(e.getMessage(), e);
             }
             if (!autoReConnect) {
                 return;
@@ -181,8 +181,9 @@ public boolean isActive() {
 
     public void close() {
         autoReConnect = false;
-        if (channel != null) {
+        if (channel != null && channel.isOpen()) {
             channel.close();
+            group.shutdownGracefully();
         }
     }
 

src/main/java/org/zhongweixian/server/websocket/WebSocketServer.java

@@ -74,10 +74,10 @@ public void start() {
                     childHandler(new WebSocketServerInitializer(connectionListener, heart, path, null));
             ChannelFuture channelFuture = bootstrap.bind().sync();
             if (channelFuture.isSuccess()) {
-                logger.info("websocket started on port :{} (websocket)", port);
+                logger.info("websocket started on port:{}, path:{}", port, path);
             }
         } catch (Exception e) {
-            logger.error("{}", e);
+            logger.error(e.getMessage(), e);
         }
     }
 

src/main/java/org/zhongweixian/server/websocket/WebSocketServerHandler.java

@@ -42,14 +42,14 @@ protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame webSocketF
                     logger.debug("received client:{}, message:{}", ctx.channel().id(), jsonObject);
                 }
                 if ("ping".equals(jsonObject.getString("cmd"))) {
-                    ctx.channel().writeAndFlush(new TextWebSocketFrame("{\"type\":\"pong\",\"sequence\":" + System.currentTimeMillis() + "}"));
+                    ctx.channel().writeAndFlush(new TextWebSocketFrame("{\"type\":\"pong\",\"status\":0,\"sequence\":" + System.currentTimeMillis() + "}"));
                     return;
                 }
                 if (jsonObject != null) {
                     listener.onMessage(ctx.channel(), textWebSocketFrame.text());
                 }
             } catch (Exception e) {
-                logger.error("解析json:{} 异常:{}", textWebSocketFrame.text(), e);
+                logger.error("解析json:{} 异常", textWebSocketFrame.text(), e);
                 JSONObject error = new JSONObject();
                 error.put("messgae", e.getMessage());
                 error.put("code", 500);