Avoid memory leakage via the thread pool

Author: mbuckton

src/main/java/io/mapsmessaging/network/protocol/impl/satellite/TaskManager.java

@@ -19,29 +19,36 @@
 
 package io.mapsmessaging.network.protocol.impl.satellite;
 
+import io.mapsmessaging.auth.AuthManager;
+
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicInteger;
 
+
+@SuppressWarnings("java:S6548") // yes it is a singleton
 public class TaskManager {
+
+  private static class Holder {
+    static final TaskManager INSTANCE = new TaskManager();
+  }
+  public static TaskManager getInstance() {
+    return Holder.INSTANCE;
+  }
+
   private static final AtomicInteger count = new AtomicInteger(1);
 
   private final ExecutorService existingPool;
   private final ScheduledExecutorService scheduler;
 
-  public TaskManager() {
+  private TaskManager() {
     this(4);
   }
 
-  public TaskManager(int poolSize) {
+  private TaskManager(int poolSize) {
     existingPool= Executors.newFixedThreadPool(4, r -> new Thread(r, "Satellite-Executors-" + count.getAndIncrement()));
     scheduler = new ScheduledThreadPoolExecutor(poolSize, ((ThreadPoolExecutor) existingPool).getThreadFactory());
   }
 
-  public void close(){
-    scheduler.shutdown();
-    existingPool.shutdown();
-  }
-
   public ScheduledFuture<?> schedule(Runnable task, long delay, TimeUnit unit) {
     return scheduler.schedule(task, delay, unit);
   }

src/main/java/io/mapsmessaging/network/protocol/impl/satellite/gateway/io/GatewayManager.java

@@ -51,7 +51,6 @@ public class GatewayManager {
   private final Map<String, RemoteDeviceInfo> knownTerminals;
   private final AtomicBoolean closed = new AtomicBoolean(false);
 
-  private TaskManager taskManager;
   private ScheduledFuture<?> scheduledFuture;
 
 
@@ -68,8 +67,7 @@ public GatewayManager(SatelliteConfigDTO satelliteConfigDTO, IncomingMessageHand
   }
 
   public void start() {
-    taskManager = new TaskManager();
-    taskManager.schedule(this::initSession, 10, TimeUnit.SECONDS);
+    TaskManager.getInstance().schedule(this::initSession, 10, TimeUnit.SECONDS);
   }
 
   public void stop(){
@@ -78,7 +76,6 @@ public void stop(){
       scheduledFuture.cancel(true);
       scheduledFuture = null;
     }
-    taskManager.close();
   }
 
   protected void initSession(){
@@ -97,15 +94,15 @@ protected void initSession(){
         for (RemoteDeviceInfo terminal : knownTerminals.values()) {
           handler.registerTerminal(terminal);
         }
-        scheduledFuture = taskManager.schedule(this::pollGateway, pollInterval, TimeUnit.SECONDS);
+        scheduledFuture = TaskManager.getInstance().schedule(this::pollGateway, pollInterval, TimeUnit.SECONDS);
       } else {
         logger.log(OGWS_FAILED_AUTHENTICATION);
       }
     }
     catch (RuntimeException| IOException e) {
       // Todo log this
       // we can try again
-      taskManager.schedule(this::initSession, 1, TimeUnit.MINUTES);
+      TaskManager.getInstance().schedule(this::initSession, 1, TimeUnit.MINUTES);
     } catch (InterruptedException e) {
       Thread.currentThread().interrupt();
     } catch (LoginException e) {
@@ -150,7 +147,7 @@ protected void pollGateway() {
       }
     }
     finally {
-      scheduledFuture = taskManager.schedule(this::pollGateway, pollInterval, TimeUnit.SECONDS);
+      scheduledFuture = TaskManager.getInstance().schedule(this::pollGateway, pollInterval, TimeUnit.SECONDS);
     }
   }
 

src/main/java/io/mapsmessaging/network/protocol/impl/satellite/gateway/protocol/SatelliteGatewayProtocol.java

@@ -60,7 +60,6 @@
 
 public class SatelliteGatewayProtocol extends Protocol {
 
-  private final TaskManager taskManager;
   private final Logger logger = LoggerFactory.getLogger(SatelliteGatewayProtocol.class);
   private final SatelliteMessageRebuilder messageRebuilder;
   private final Session session;
@@ -88,7 +87,6 @@ public SatelliteGatewayProtocol(@NonNull @NotNull EndPoint endPoint, @NotNull @N
     pendingMessages.set(new LinkedHashMap<>());
     priorityMessages = new AtomicReference<>();
     priorityMessages.set(new LinkedHashMap<>());
-    taskManager = new TaskManager();
     String primeId = ((SatelliteEndPoint) endPoint).getTerminalInfo().getUniqueId();
     SatelliteConfigDTO config = (SatelliteConfigDTO) protocolConfig;
     sendHighPriorityEvents = config.isSendHighPriorityMessages();
@@ -151,7 +149,7 @@ public SatelliteGatewayProtocol(@NonNull @NotNull EndPoint endPoint, @NotNull @N
     }
     ((SatelliteEndPoint) endPoint).unmute();
     nextOutgoingTime = System.currentTimeMillis() + outgoingPollInterval;
-    scheduledFuture = taskManager.schedule(this::processOutstandingMessages, 15, TimeUnit.SECONDS);
+    scheduledFuture = TaskManager.getInstance().schedule(this::processOutstandingMessages, 15, TimeUnit.SECONDS);
   }
 
   private String parsePath(String path, String defaultValue, String primeId, String mailboxId){
@@ -172,7 +170,6 @@ public void close() throws IOException {
       closed = true;
       SessionManager.getInstance().close(session, false);
       super.close();
-      taskManager.close();
     }
   }
 
@@ -321,7 +318,7 @@ private void processOutstandingMessages() {
         // Log this
       }
     } finally {
-      scheduledFuture = taskManager.schedule(this::processOutstandingMessages, 15, TimeUnit.SECONDS);
+      scheduledFuture = TaskManager.getInstance().schedule(this::processOutstandingMessages, 15, TimeUnit.SECONDS);
     }
   }
 

src/main/java/io/mapsmessaging/network/protocol/impl/satellite/modem/device/Modem.java

@@ -61,7 +61,6 @@ public class Modem {
   private final Queue<Command> commandQueue = new ArrayDeque<>();
   private final StringBuilder responseBuffer = new StringBuilder();
   private final long modemTimeout;
-  private final TaskManager taskManager;
 
   @Getter
   private final ModemStreamHandler streamHandler;
@@ -78,18 +77,17 @@ public class Modem {
   @Setter
   private boolean oneShotResponse;
 
-  public Modem(Consumer<Packet> packetSender, long modemTimeout, ModemStreamHandler streamHandler, TaskManager taskManager) {
+  public Modem(Consumer<Packet> packetSender, long modemTimeout, ModemStreamHandler streamHandler) {
     this.streamHandler = streamHandler;
     this.packetSender = packetSender;
-    this.taskManager = taskManager;
     oneShotResponse = false;
     currentHandler = new TextResponseHandler(this::handleLine);
     if (modemTimeout < 10000 || modemTimeout > 120000) {
       modemTimeout = 15000;
     }
     this.modemTimeout = modemTimeout;
 
-    future = taskManager.schedule(this::scanTimeouts, modemTimeout, TimeUnit.MILLISECONDS);
+    future = TaskManager.getInstance().schedule(this::scanTimeouts, modemTimeout, TimeUnit.MILLISECONDS);
   }
 
   public void close() {
@@ -111,7 +109,7 @@ private void scanTimeouts() {
         currentCommand.future.completeExceptionally(new IOException("Modem has been closed"));
       }
     } finally {
-      future = taskManager.schedule(this::scanTimeouts, modemTimeout, TimeUnit.MILLISECONDS);
+      future = TaskManager.getInstance().schedule(this::scanTimeouts, modemTimeout, TimeUnit.MILLISECONDS);
     }
   }
 

src/main/java/io/mapsmessaging/network/protocol/impl/satellite/modem/protocol/StoGiProtocol.java

@@ -76,7 +76,6 @@ public class StoGiProtocol extends Protocol implements Consumer<Packet> {
 
   private static final String STOGI = "stogi";
 
-  private final TaskManager taskManager;
   private final Logger logger = LoggerFactory.getLogger(StoGiProtocol.class);
   private final Session session;
   private final SelectorTask selectorTask;
@@ -112,7 +111,6 @@ public StoGiProtocol(EndPoint endPoint, Packet packet) throws LoginException, IO
     super(endPoint, endPoint.getConfig().getProtocolConfig(STOGI));
     satelliteOnline = false;
     satelliteOnlineCount = 0;
-    taskManager = new TaskManager();
     satelliteMessageRebuilder = new SatelliteMessageRebuilder();
     messageId = 0;
     currentList = new ArrayList<>();
@@ -144,7 +142,7 @@ public StoGiProtocol(EndPoint endPoint, Packet packet) throws LoginException, IO
 
     sinNumber = modemConfig.getSinNumber();
     messageLifeTime = modemConfig.getMessageLifeTimeInMinutes();
-    modem = new Modem(this, modemConfig.getModemResponseTimeout(), streamHandler, taskManager);
+    modem = new Modem(this, modemConfig.getModemResponseTimeout(), streamHandler);
 
     if(!modemConfig.getSharedSecret().trim().isEmpty()) {
       cipherManager = new CipherManager(modemConfig.getSharedSecret().getBytes());
@@ -174,7 +172,7 @@ public StoGiProtocol(EndPoint endPoint, Packet packet) throws LoginException, IO
     }
     lastOutgoingMessagePollInterval = System.currentTimeMillis();
     statsManager = new StatsManager(modem, locationPollInterval, destination);
-    scheduledFuture = taskManager.schedule(this::pollModemForMessages, messagePoll, TimeUnit.MILLISECONDS);
+    scheduledFuture = TaskManager.getInstance().schedule(this::pollModemForMessages, messagePoll, TimeUnit.MILLISECONDS);
     logger.log(STOGI_STARTED_SESSION, modem.getModemProtocol().getType(), messagePoll,outgoingMessagePollInterval );
   }
 
@@ -212,7 +210,7 @@ private void pollModemForMessages() {
       logger.log(STOGI_POLL_RAISED_EXCEPTION, th);
     } finally {
       poll = hasOutgoing?500:poll;
-      scheduledFuture = taskManager.schedule(this::pollModemForMessages, poll, TimeUnit.MILLISECONDS);
+      scheduledFuture = TaskManager.getInstance().schedule(this::pollModemForMessages, poll, TimeUnit.MILLISECONDS);
       startTime = System.currentTimeMillis() - startTime;
       logger.log(STOGI_POLL_FOR_ACTIONS, startTime, poll);
     }
@@ -228,7 +226,6 @@ public void close() throws IOException {
       modem.close();
     }
     super.close();
-    taskManager.close();
   }
 
   private Session setupSession() throws LoginException, IOException {