Added proper shutdown functionality of unused inputs, outputs and their connectors. Updated API and framework.

Author: sebinside

src/main/scala/org/codeoverflow/chatoverflow/Launcher.scala

@@ -11,6 +11,7 @@ object Launcher extends WithLogger {
 
   var server: Option[Server] = None
   var pluginDataPath: String = "data"
+  private var chatOverflow: Option[ChatOverflow] = None
 
   /**
     * Software entry point.
@@ -24,6 +25,7 @@ object Launcher extends WithLogger {
       // The complete project visible trough one single instance
       val chatOverflow = new ChatOverflow(config.pluginFolderPath,
         config.configFolderPath, config.requirementPackage, config.pluginLogOutputOnConsole)
+      this.chatOverflow = Some(chatOverflow)
 
       // Initialize chat overflow
       chatOverflow.init()
@@ -68,7 +70,25 @@ object Launcher extends WithLogger {
     * Shuts down the framework.
     */
   def exit(): Unit = {
-    logger info "Shutting down Chat Overflow."
+    logger debug "Shutting down Chat Overflow."
+    if (chatOverflow.isDefined) {
+      logger debug "Trying to stop all running instances..."
+      chatOverflow.get.pluginInstanceRegistry.getAllPluginInstances.
+        filter(_.isRunning).foreach(_.stopPlease())
+
+      for (i <- 1 to 5) {
+        if (chatOverflow.get.pluginInstanceRegistry.getAllPluginInstances.
+          exists(_.isRunning)) {
+          logger info s"Check $i of 5. Still waiting on: ${
+            chatOverflow.get.pluginInstanceRegistry.getAllPluginInstances.
+              filter(_.isRunning).map(_.instanceName).mkString(", ")
+          }"
+          Thread.sleep(1000)
+        }
+      }
+    }
+
+    logger info "Bye Bye. Stay minzig!"
     System.exit(0)
   }
 }

src/main/scala/org/codeoverflow/chatoverflow/connector/Connector.scala

@@ -25,6 +25,8 @@ abstract class Connector(val sourceIdentifier: String) extends WithLogger {
   protected var optionalCredentialKeys: List[String]
   protected var running = false
 
+  private var instanceCount = 0
+
   def getCredentials: Option[Credentials] = this.credentials
 
   /**
@@ -76,6 +78,7 @@ abstract class Connector(val sourceIdentifier: String) extends WithLogger {
     // Check if running
     if (running) {
       logger info s"Unable to start $connectorSourceAndType. Already running!"
+      changeInstanceCount(1)
       true
     } else {
 
@@ -101,6 +104,7 @@ abstract class Connector(val sourceIdentifier: String) extends WithLogger {
 
           if (start()) {
             logger info s"Started $connectorSourceAndType."
+            changeInstanceCount(1)
             running = true
             true
           } else {
@@ -127,15 +131,29 @@ abstract class Connector(val sourceIdentifier: String) extends WithLogger {
   /**
     * Shuts down the connector by calling the stop method.
     */
-  def shutdown(): Unit = {
-    if (stop()) {
-      running = false
-      logger info s"Stopped $connectorSourceAndType."
+  def shutdown(): Boolean = {
+    changeInstanceCount(-1)
+
+    if (instanceCount <= 0) {
+      logger info "Instance count is zero. Connector is no longer needed and shut down. RIP."
+      if (stop()) {
+        running = false
+        logger info s"Stopped $connectorSourceAndType."
+        true
+      } else {
+        logger warn s"Unable to shutdown $connectorSourceAndType."
+        false
+      }
     } else {
-      logger warn s"Unable to shutdown $connectorSourceAndType."
+      true // Return true to keep transparency
     }
   }
 
+  private def changeInstanceCount(delta: Int): Unit = {
+    instanceCount += delta
+    logger info s"Instance count: $instanceCount"
+  }
+
   /**
     * Returns the unique type string of the implemented connector.
     *

src/main/scala/org/codeoverflow/chatoverflow/instance/PluginInstance.scala

@@ -73,12 +73,6 @@ class PluginInstance(val instanceName: String, pluginType: PluginType) extends W
     new PluginManagerStub()
   }
 
-  /**
-    * Returns, if the plugin instance is created. This should be always the case when a plugin instance
-    * is added to the registry.
-    */
-  private def isCreated: Boolean = plugin.isDefined
-
   /**
     * Creates a new thread and tries to start the plugin.
     * Make sure to set the requirements first!
@@ -183,6 +177,20 @@ class PluginInstance(val instanceName: String, pluginType: PluginType) extends W
                   case e: AbstractMethodError => logger.error(s"Plugin '$instanceName' just crashed. Looks like a plugin version error.", e)
                   case e: Exception => logger.error(s"Plugin '$instanceName' just had an exception. Might be a plugin implementation fault.", e)
                   case e: Throwable => logger.error(s"Plugin '$instanceName' just crashed. We don't know whats going up here!", e)
+
+                } finally {
+
+                  // If the plugin ended or crashed, try to shut down the connectors
+                  val inputRequirements = getRequirements.getInputRequirements.toArray
+                  inputRequirements.foreach(requirement => {
+                    requirement.asInstanceOf[Requirement[Input]].get().shutdown()
+                  })
+
+                  val outputRequirements = getRequirements.getOutputRequirements.toArray
+                  outputRequirements.foreach(requirement => {
+                    requirement.asInstanceOf[Requirement[Output]].get().shutdown()
+                  })
+
                 }
               })
               instanceThread.start()
@@ -198,6 +206,12 @@ class PluginInstance(val instanceName: String, pluginType: PluginType) extends W
     }
   }
 
+  /**
+    * Returns, if the plugin instance is created. This should be always the case when a plugin instance
+    * is added to the registry.
+    */
+  private def isCreated: Boolean = plugin.isDefined
+
   /**
     * Returns the requirements object of the specific plugin instance.
     *

src/main/scala/org/codeoverflow/chatoverflow/requirement/InputImpl.scala

@@ -9,9 +9,9 @@ import scala.reflect.ClassTag
 abstract class InputImpl[C <: Connector](implicit ct: ClassTag[C]) extends Connection[C] with Input with WithLogger {
 
   /**
-    * Inits this connection, checks if teh source connector is defined, and can be inited, then calls start
+    * Init this connection, checks if the source connector is defined, and can be inited, then calls start
     *
-    * @return if this input could be successfully inited
+    * @return if this input could be successfully initialized
     */
   override def init(): Boolean = {
     if (sourceConnector.isDefined) {
@@ -24,6 +24,20 @@ abstract class InputImpl[C <: Connector](implicit ct: ClassTag[C]) extends Conne
     }
   }
 
+  /**
+    * Shuts down the connection by calling stop on the input and requesting shutdown on the connector.
+    *
+    * @return true if both parties tried to shut down
+    */
+  override def shutdown(): Boolean = {
+    if (sourceConnector.isDefined) {
+      stop() & sourceConnector.get.shutdown()
+    } else {
+      logger warn "Source connector not set."
+      false
+    }
+  }
+
 
   /**
     * Start the input, called after source connector did init
@@ -32,6 +46,13 @@ abstract class InputImpl[C <: Connector](implicit ct: ClassTag[C]) extends Conne
     */
   def start(): Boolean
 
+  /**
+    * Stops the input, called before source connector will shutdown
+    *
+    * @return true if stopping was successful
+    */
+  def stop(): Boolean
+
 
   /**
     * Serializes this object into a string to save it to a config

src/main/scala/org/codeoverflow/chatoverflow/requirement/OutputImpl.scala

@@ -9,9 +9,9 @@ import scala.reflect.ClassTag
 abstract class OutputImpl[C <: Connector](implicit ct: ClassTag[C]) extends Connection[C] with Output with WithLogger {
 
   /**
-    * Inits this connection, checks if teh source connector is defined, and can be inited, then calls start
+    * Init this connection, checks if teh source connector is defined, and can be inited, then calls start
     *
-    * @return if this input could be successfully inited
+    * @return if this input could be successfully initialized
     */
   override def init(): Boolean = {
     if (sourceConnector.isDefined) {
@@ -24,6 +24,19 @@ abstract class OutputImpl[C <: Connector](implicit ct: ClassTag[C]) extends Conn
     }
   }
 
+  /**
+    * Shuts down the connection by calling stop on the output and requesting shutdown on the connector.
+    *
+    * @return true if both parties tried to shut down
+    */
+  override def shutdown(): Boolean = {
+    if (sourceConnector.isDefined) {
+      stop() & sourceConnector.get.shutdown()
+    } else {
+      logger warn "Source connector not set."
+      false
+    }
+  }
 
   /**
     * Start the input, called after source connector did init
@@ -32,6 +45,13 @@ abstract class OutputImpl[C <: Connector](implicit ct: ClassTag[C]) extends Conn
     */
   def start(): Boolean
 
+  /**
+    * Stops the output, called before source connector will shutdown
+    *
+    * @return true if stopping was successful
+    */
+  def stop(): Boolean
+
 
   /**
     * Serializes this object into a string to save it to a config

src/main/scala/org/codeoverflow/chatoverflow/requirement/service/discord/impl/DiscordChatInputImpl.scala

@@ -238,6 +238,13 @@ class DiscordChatInputImpl extends InputImpl[DiscordChatConnector] with DiscordC
       }
     }
   }
+
+  /**
+    * Stops the input, called before source connector will shutdown
+    *
+    * @return true if stopping was successful
+    */
+  override def stop(): Boolean = true
 }
 
 object DiscordChatInputImpl {

src/main/scala/org/codeoverflow/chatoverflow/requirement/service/discord/impl/DiscordChatOutputImpl.scala

@@ -52,4 +52,11 @@ class DiscordChatOutputImpl extends OutputImpl[DiscordChatConnector] with Discor
   override def sendFile(file: String): Unit = sourceConnector.get.sendFile(getChannelId, file)
 
   override def sendFile(file: String, message: String): Unit = sourceConnector.get.sendFile(getChannelId, file, Some(message))
+
+  /**
+    * Stops the output, called before source connector will shutdown
+    *
+    * @return true if stopping was successful
+    */
+  override def stop(): Boolean = true
 }

src/main/scala/org/codeoverflow/chatoverflow/requirement/service/file/impl/FileInputImpl.scala

@@ -14,22 +14,25 @@ import org.codeoverflow.chatoverflow.requirement.service.file.FileConnector
 @Impl(impl = classOf[FileInput], connector = classOf[FileConnector])
 class FileInputImpl extends InputImpl[FileConnector] with FileInput with WithLogger {
 
-  override def init(): Boolean = {
-    sourceConnector.get.init()
-  }
-
   override def getFile(pathInResources: String): Optional[String] = Optional.ofNullable(sourceConnector.get.getFile(pathInResources).orNull)
 
   override def getBinaryFile(pathInResources: String): Optional[Array[Byte]] = Optional.ofNullable(sourceConnector.get.getBinaryFile(pathInResources).orNull)
 
   override def getImage(pathInResources: String): Optional[BufferedImage] = {
     val data = sourceConnector.get.getBinaryFile(pathInResources)
-    if (!data.isDefined) {
+    if (data.isEmpty) {
       None
     }
     val bis = new ByteArrayInputStream(data.get)
     Optional.of(ImageIO.read(bis))
   }
 
   override def start(): Boolean = true
+
+  /**
+    * Stops the input, called before source connector will shutdown
+    *
+    * @return true if stopping was successful
+    */
+  override def stop(): Boolean = true
 }

src/main/scala/org/codeoverflow/chatoverflow/requirement/service/file/impl/FileOutputImpl.scala

@@ -13,10 +13,6 @@ import org.codeoverflow.chatoverflow.requirement.service.file.FileConnector
 @Impl(impl = classOf[FileOutput], connector = classOf[FileConnector])
 class FileOutputImpl extends OutputImpl[FileConnector] with FileOutput with WithLogger {
 
-  override def init() = {
-    sourceConnector.get.init()
-  }
-
   override def saveFile(content: String, pathInResources: String): Boolean = {
     sourceConnector.get.saveFile(pathInResources, content)
   }
@@ -45,4 +41,10 @@ class FileOutputImpl extends OutputImpl[FileConnector] with FileOutput with With
 
   override def start() = true
 
+  /**
+    * Stops the output, called before source connector will shutdown
+    *
+    * @return true if stopping was successful
+    */
+  override def stop(): Boolean = true
 }

src/main/scala/org/codeoverflow/chatoverflow/requirement/service/mockup/impl/MockUpChatInputImpl.scala

@@ -7,15 +7,15 @@ import org.codeoverflow.chatoverflow.WithLogger
 import org.codeoverflow.chatoverflow.api.io.dto.chat.{Channel, ChatEmoticon, ChatMessage, ChatMessageAuthor}
 import org.codeoverflow.chatoverflow.api.io.input.chat.MockUpChatInput
 import org.codeoverflow.chatoverflow.registry.Impl
-import org.codeoverflow.chatoverflow.requirement.Connection
+import org.codeoverflow.chatoverflow.requirement.InputImpl
 import org.codeoverflow.chatoverflow.requirement.service.mockup.MockUpChatConnector
 
 import scala.collection.JavaConverters._
 import scala.collection.mutable.ListBuffer
 
 @Deprecated
 @Impl(impl = classOf[MockUpChatInput], connector = classOf[MockUpChatConnector])
-class MockUpChatInputImpl extends Connection[MockUpChatConnector] with MockUpChatInput with WithLogger {
+class MockUpChatInputImpl extends InputImpl[MockUpChatConnector] with MockUpChatInput with WithLogger {
 
   private val messages: ListBuffer[ChatMessage[ChatMessageAuthor, Channel, ChatEmoticon]] = ListBuffer[ChatMessage[ChatMessageAuthor, Channel, ChatEmoticon]]()
   private val privateMessages: ListBuffer[ChatMessage[ChatMessageAuthor, Channel, ChatEmoticon]] = ListBuffer[ChatMessage[ChatMessageAuthor, Channel, ChatEmoticon]]()
@@ -40,22 +40,26 @@ class MockUpChatInputImpl extends Connection[MockUpChatConnector] with MockUpCha
 
   override def registerPrivateMessageHandler(handler: Consumer[ChatMessage[ChatMessageAuthor, Channel, ChatEmoticon]]): Unit = privateMessageHandler += handler
 
-  override def init(): Boolean = {
-    if (sourceConnector.isDefined) {
-      sourceConnector.get.addMessageEventListener(onMessage)
-      sourceConnector.get.init()
-    } else {
-      logger warn "Source connector not set."
-      false
-    }
-  }
+  override def serialize(): String = getSourceIdentifier
+
+  override def deserialize(value: String): Unit = setSourceConnector(value)
+
+  /**
+    * Start the input, called after source connector did init
+    *
+    * @return true if starting the input was successful, false if some problems occurred
+    */
+  override def start(): Boolean = true
 
   private def onMessage(msg: ChatMessage[ChatMessageAuthor, Channel, ChatEmoticon]): Unit = {
     messageHandler.foreach(consumer => consumer.accept(msg))
     messages += msg
   }
 
-  override def serialize(): String = getSourceIdentifier
-
-  override def deserialize(value: String): Unit = setSourceConnector(value)
+  /**
+    * Stops the input, called before source connector will shutdown
+    *
+    * @return true if stopping was successful
+    */
+  override def stop(): Boolean = true
 }