Bridge:fix: бесконечный реконнект

bridge/build.gradle

@@ -1,5 +1,5 @@
 group = 'asys'
-version = '0.5.2-SNAPSHOT'
+version = '0.5.3-SNAPSHOT'
 
 repositories {
     maven { url 'https://hub.spigotmc.org/nexus/content/groups/public/' }

bridge/src/main/java/asys/bridge/bukkit/BridgePlugin.java

@@ -13,18 +13,13 @@ import org.bukkit.command.Command;
 import org.bukkit.command.CommandSender;
 import org.bukkit.plugin.java.JavaPlugin;
 
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-
 public class BridgePlugin extends JavaPlugin {
     public static BridgePlugin INSTANCE;
     private Client client;
-    private ScheduledExecutorService ses;
+    private TaskTicker connectTicker, pingTicker;
-    private ScheduledFuture<?> sesFuture, sesPingFuture;
     private int tryConnect = 0;
     private BridgeLoggerAppender loggerAppender;
+    private boolean needReconnect = true;
 
     @Override
     public void onLoad() {
@@ -44,10 +39,8 @@ public class BridgePlugin extends JavaPlugin {
 
     @Override
     public void onDisable() {
-        if (sesFuture != null) {
+        setNeedReconnect(false);
-            sesFuture.cancel(false);
+        stopReconnect();
-        }
-
         stopPing();
 
         if (client.isConnected()) {
@@ -66,8 +59,8 @@ public class BridgePlugin extends JavaPlugin {
 
     public void startReconnect() {
         client = new Client();
-        ses = Executors.newScheduledThreadPool(2);
+        connectTicker = new TaskTicker().setStepTimeMs(5000L);
-        sesFuture = ses.scheduleAtFixedRate(() -> {
+        connectTicker.setTask(() -> {
             getLogger().info(String.format("Connect(%d) to ASys...", ++tryConnect));
             client.connect(getConfig().getString("host"), getConfig().getInt("port"));
             if (client.isConnected()) {
@@ -76,20 +69,20 @@ public class BridgePlugin extends JavaPlugin {
                 getLogger().warning(
                         String.format("Connection(%d) fail. Try reconnect...", tryConnect));
             }
-        }, 0L, 5L, TimeUnit.SECONDS);
+        }).start();
     }
 
     public void stopReconnect() {
-        if (sesFuture != null) {
+        if (connectTicker != null) {
-            sesFuture.cancel(false);
+            connectTicker.stop();
-            sesFuture = null;
             tryConnect = 0;
         }
     }
 
     public void startPing(Channel channel) {
         getLoggerAppender().setChannel(channel);
-        sesPingFuture = ses.scheduleAtFixedRate(() -> {
+        pingTicker = new TaskTicker().setStepTimeMs(5000L);
+        pingTicker.setTask(() -> {
             channel.write(new CS_Ping(
                     System.currentTimeMillis(),
                     20.0D, //FIXME
@@ -105,17 +98,25 @@ public class BridgePlugin extends JavaPlugin {
                 getLogger().warning("Try reconnect...");
                 startReconnect();
             }
-        }, 0L, 5L, TimeUnit.SECONDS);
+        }).start();
     }
 
     public void stopPing() {
         getLoggerAppender().setChannel(null);
-        if (sesPingFuture != null) {
+        if (pingTicker != null) {
-            sesPingFuture.cancel(false);
+            pingTicker.stop();
         }
     }
 
     public BridgeLoggerAppender getLoggerAppender() {
         return loggerAppender;
     }
+
+    public boolean isNeedReconnect() {
+        return needReconnect;
+    }
+
+    public void setNeedReconnect(boolean needReconnect) {
+        this.needReconnect = needReconnect;
+    }
 }

bridge/src/main/java/asys/bridge/bukkit/TaskTicker.java

@@ -0,0 +1,48 @@
+/*
+ * DmitriyMX <d.mihailov@samson-rus.com>
+ * 2017-05-18
+ */
+package asys.bridge.bukkit;
+
+public class TaskTicker implements Runnable {
+    private Runnable task;
+    private long stepTimeMs = 1000L;
+    private Thread thread;
+    private boolean loop = false;
+
+    TaskTicker setTask(Runnable task) {
+        this.task = task;
+        return this;
+    }
+
+    TaskTicker setStepTimeMs(long stepTimeMs) {
+        this.stepTimeMs = stepTimeMs;
+        return this;
+    }
+
+    void start() {
+        thread = new Thread(this, "TaskTicker");
+        loop = true;
+        thread.start();
+    }
+
+    void stop() {
+        loop = false;
+        if (thread != null) {
+            thread.interrupt();
+        }
+    }
+
+    @Override
+    public void run() {
+        while (loop || !Thread.currentThread().isInterrupted()) {
+            task.run();
+
+            try {
+                Thread.sleep(stepTimeMs);
+            } catch (InterruptedException e) {
+                break;
+            }
+        }
+    }
+}

bridge/src/main/java/asys/bridge/client/ClientPacketHandler.java

@@ -54,11 +54,12 @@ public class ClientPacketHandler extends ChannelInboundHandlerAdapter implements
     @Override
     public void channelInactive(ChannelHandlerContext context) throws Exception {
         if (BridgePlugin.INSTANCE != null) {
-            BridgePlugin.INSTANCE.getLogger().warning("Lost connection!");
             BridgePlugin.INSTANCE.stopPing();
-            BridgePlugin.INSTANCE.getLogger().warning("Try reconnect...");
+            if (BridgePlugin.INSTANCE.isNeedReconnect()) {
+                BridgePlugin.INSTANCE.getLogger().warning("Lost connection! Try reconnect...");
                 BridgePlugin.INSTANCE.startReconnect();
             }
+        }
         context.channel().attr(KNOWN_PACKETS).remove();
         context.channel().attr(KNOWN_HANDLERS).remove();
         super.channelInactive(context);
@@ -71,6 +72,7 @@ public class ClientPacketHandler extends ChannelInboundHandlerAdapter implements
         if (pkt.getErrorCode() != 0) {
             BridgePlugin.INSTANCE.getLogger().severe(
                     String.format("Handshake: #%d %s", pkt.getErrorCode(), pkt.getMessage()));
+            BridgePlugin.INSTANCE.setNeedReconnect(false);
         } else {
             context.channel().attr(KNOWN_PACKETS).set(pingPackets);
             BridgePlugin.INSTANCE.getLogger().info("Handshake: OK");