Cherry pick fix for #89 to develop

src/main/scala/org/codeoverflow/chatoverflow/requirement/impl/EventInputImpl.scala

@@ -39,4 +39,15 @@ abstract class EventInputImpl[T <: Event, C <: Connector](implicit ctc: ClassTag
     handlers.filter(handler => handler.clazz == cts.runtimeClass)
       .foreach(handler => handler.consumer.asInstanceOf[Consumer[S]].accept(event))
   }
+
+  override def shutdown(): Boolean = {
+    if (sourceConnector.isDefined) {
+      val stopped = stop()
+      handlers.clear()
+      stopped & sourceConnector.get.shutdown()
+    } else {
+      logger warn "Source connector not set."
+      false
+    }
+  }
 }

src/main/scala/org/codeoverflow/chatoverflow/requirement/service/discord/DiscordChatConnector.scala

@@ -47,6 +47,21 @@ class DiscordChatConnector(override val sourceIdentifier: String) extends Connec
   def addReactionDelEventListener(listener: MessageReactionRemoveEvent => Unit): Unit =
     discordChatListener.addReactionDelEventListener(listener)
 
+  def removeMessageReceivedListener(listener: MessageReceivedEvent => Unit): Unit =
+    discordChatListener.removeMessageReceivedListener(listener)
+
+  def removeMessageUpdateListener(listener: MessageUpdateEvent => Unit): Unit =
+    discordChatListener.removeMessageUpdateEventListener(listener)
+
+  def removeMessageDeleteListener(listener: MessageDeleteEvent => Unit): Unit =
+    discordChatListener.removeMessageDeleteEventListener(listener)
+
+  def removeReactionAddEventListener(listener: MessageReactionAddEvent => Unit): Unit =
+    discordChatListener.removeReactionAddEventListener(listener)
+
+  def removeReactionDelEventListener(listener: MessageReactionRemoveEvent => Unit): Unit =
+    discordChatListener.removeReactionDelEventListener(listener)
+
   /**
     * Connects to discord
     */

src/main/scala/org/codeoverflow/chatoverflow/requirement/service/discord/DiscordChatListener.scala

@@ -32,6 +32,16 @@ class DiscordChatListener extends EventListener {
 
   def addReactionDelEventListener(listener: MessageReactionRemoveEvent => Unit): Unit = reactionDelEventListener += listener
 
+  def removeMessageReceivedListener(listener: MessageReceivedEvent => Unit): Unit = messageEventListener -= listener
+
+  def removeMessageUpdateEventListener(listener: MessageUpdateEvent => Unit): Unit = messageUpdateEventListener -= listener
+
+  def removeMessageDeleteEventListener(listener: MessageDeleteEvent => Unit): Unit = messageDeleteEventListener -= listener
+
+  def removeReactionAddEventListener(listener: MessageReactionAddEvent => Unit): Unit = reactionAddEventListener -= listener
+
+  def removeReactionDelEventListener(listener: MessageReactionRemoveEvent => Unit): Unit = reactionDelEventListener -= listener
+
   override def onEvent(event: Event): Unit = {
     event match {
       case receivedEvent: MessageReceivedEvent => messageEventListener.foreach(listener => listener(receivedEvent))

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

@@ -34,12 +34,18 @@ class DiscordChatInputImpl extends EventInputImpl[DiscordEvent, DiscordChatConne
   private val privateMessages = ListBuffer[DiscordChatMessage]()
   private var channelId: Option[String] = None
 
+  private val onMessageFn = onMessage _
+  private val onMessageUpdateFn = onMessageUpdate _
+  private val onMessageDeleteFn = onMessageDelete _
+  private val onReactionAddedFn = onReactionAdded _
+  private val onReactionRemovedFn = onReactionRemoved _
+
   override def start(): Boolean = {
-    sourceConnector.get.addMessageReceivedListener(onMessage)
-    sourceConnector.get.addMessageUpdateListener(onMessageUpdate)
-    sourceConnector.get.addMessageDeleteListener(onMessageDelete)
-    sourceConnector.get.addReactionAddEventListener(onReactionAdded)
-    sourceConnector.get.addReactionDelEventListener(onReactionRemoved)
+    sourceConnector.get.addMessageReceivedListener(onMessageFn)
+    sourceConnector.get.addMessageUpdateListener(onMessageUpdateFn)
+    sourceConnector.get.addMessageDeleteListener(onMessageDeleteFn)
+    sourceConnector.get.addReactionAddEventListener(onReactionAddedFn)
+    sourceConnector.get.addReactionDelEventListener(onReactionRemovedFn)
     true
   }
 
@@ -81,7 +87,14 @@ class DiscordChatInputImpl extends EventInputImpl[DiscordEvent, DiscordChatConne
     *
     * @return true if stopping was successful
     */
-  override def stop(): Boolean = true
+  override def stop(): Boolean = {
+    sourceConnector.get.removeMessageReceivedListener(onMessageFn)
+    sourceConnector.get.removeMessageUpdateListener(onMessageUpdateFn)
+    sourceConnector.get.removeMessageDeleteListener(onMessageDeleteFn)
+    sourceConnector.get.removeReactionAddEventListener(onReactionAddedFn)
+    sourceConnector.get.removeReactionDelEventListener(onReactionRemovedFn)
+    true
+  }
 
   /**
     * Listens for received messages, parses the data, adds them to the buffer and handles them over to the correct handler

src/main/scala/org/codeoverflow/chatoverflow/requirement/service/serial/SerialConnector.scala

@@ -2,10 +2,12 @@ package org.codeoverflow.chatoverflow.requirement.service.serial
 
 import java.io.{InputStream, PrintStream}
 
-import com.fazecast.jSerialComm.{SerialPort, SerialPortInvalidPortException}
+import com.fazecast.jSerialComm.{SerialPort, SerialPortEvent, SerialPortInvalidPortException}
 import org.codeoverflow.chatoverflow.WithLogger
 import org.codeoverflow.chatoverflow.connector.Connector
 
+import scala.collection.mutable
+
 /**
   * The serial connector allows to communicate with a device connected to the pcs serial port (like an Arduino)
   *
@@ -19,6 +21,7 @@ class SerialConnector(override val sourceIdentifier: String) extends Connector(s
   private var serialPort: Option[SerialPort] = None
   private var out: Option[PrintStream] = None
   private var in: Option[InputStream] = None
+  private val inputListeners: mutable.Map[Array[Byte] => Unit, SerialPortEvent => Unit] = mutable.Map()
 
   /**
     * @throws java.lang.IllegalStateException if the serial port is not available yet
@@ -49,11 +52,20 @@ class SerialConnector(override val sourceIdentifier: String) extends Connector(s
   @throws(classOf[IllegalStateException])
   def addInputListener(listener: Array[Byte] => Unit): Unit = {
     if (serialPort.isEmpty) throw new IllegalStateException("Serial port is not available yet")
-    serialPortInputListener.addDataAvailableListener(_ => {
+    val l: SerialPortEvent => Unit = _ => {
       val buffer = new Array[Byte](serialPort.get.bytesAvailable())
       serialPort.get.readBytes(buffer, buffer.length)
       listener(buffer)
-    })
+    }
+    inputListeners += (listener -> l)
+    serialPortInputListener.addDataAvailableListener(l)
+  }
+
+  def removeInputListener(listener: Array[Byte] => Unit): Unit = {
+    inputListeners remove listener match {
+      case Some(l) => serialPortInputListener.removeDataAvailableListener(l)
+      case _ => //listener not found, do nothing
+    }
   }
 
   /**
@@ -93,6 +105,7 @@ class SerialConnector(override val sourceIdentifier: String) extends Connector(s
     * Closes the connection with the port
     */
   override def stop(): Boolean = {
+
     serialPort.foreach(_.closePort())
     true
   }

src/main/scala/org/codeoverflow/chatoverflow/requirement/service/serial/SerialPortInputListener.scala

@@ -17,4 +17,6 @@ class SerialPortInputListener extends SerialPortDataListener {
   }
 
   def addDataAvailableListener(listener: SerialPortEvent => Unit): Unit = listeners += listener
+
+  def removeDataAvailableListener(listener: SerialPortEvent => Unit): Unit = listeners += listener
 }

src/main/scala/org/codeoverflow/chatoverflow/requirement/service/serial/impl/SerialInputImpl.scala

@@ -12,17 +12,24 @@ import org.codeoverflow.chatoverflow.requirement.service.serial.SerialConnector
 @Impl(impl = classOf[SerialInput], connector = classOf[SerialConnector])
 class SerialInputImpl extends EventInputImpl[SerialEvent, SerialConnector] with SerialInput with WithLogger {
 
+  private val onInputFn = onInput _
+
   override def start(): Boolean = {
-    sourceConnector.get.addInputListener(bytes => call(new SerialDataAvailableEvent(bytes)))
+    sourceConnector.get.addInputListener(onInputFn)
     true
   }
 
+  private def onInput(bytes: Array[Byte]): Unit = call(new SerialDataAvailableEvent(bytes))
+
   override def getInputStream: InputStream = sourceConnector.get.getInputStream
 
   /**
     * Stops the input, called before source connector will shutdown
     *
     * @return true if stopping was successful
     */
-  override def stop(): Boolean = true
+  override def stop(): Boolean = {
+    sourceConnector.get.removeInputListener(onInputFn)
+    true
+  }
 }

src/main/scala/org/codeoverflow/chatoverflow/requirement/service/tipeeestream/TipeeestreamConnector.scala

@@ -86,6 +86,12 @@ class TipeeestreamConnector(override val sourceIdentifier: String) extends Conne
 
   def addFollowEventListener(listener: JSONObject => Unit): Unit = tipeeeStreamListener.addFollowEventListener(listener)
 
+  def removeSubscriptionEventListener(listener: JSONObject => Unit): Unit = tipeeeStreamListener.removeSubscriptionEventListener(listener)
+
+  def removeDonationEventListener(listener: JSONObject => Unit): Unit = tipeeeStreamListener.removeDonationEventListener(listener)
+
+  def removeFollowEventListener(listener: JSONObject => Unit): Unit = tipeeeStreamListener.removeFollowEventListener(listener)
+
   override def stop(): Boolean = {
     socket.foreach(_.close())
     true

src/main/scala/org/codeoverflow/chatoverflow/requirement/service/tipeeestream/TipeeestreamListener.scala

@@ -23,6 +23,18 @@ class TipeeestreamListener {
     followEventListeners += listener
   }
 
+  def removeSubscriptionEventListener(listener: JSONObject => Unit): Unit = {
+    subscriptionEventListeners -= listener
+  }
+
+  def removeDonationEventListener(listener: JSONObject => Unit): Unit = {
+    donationEventListeners -= listener
+  }
+
+  def removeFollowEventListener(listener: JSONObject => Unit): Unit = {
+    followEventListeners -= listener
+  }
+
   def onSocketEvent(objects : Array[AnyRef]) : Unit = {
     val json: JSONObject = objects(0).asInstanceOf[JSONObject]
     val event: JSONObject = json.getJSONObject("event")

src/main/scala/org/codeoverflow/chatoverflow/requirement/service/tipeeestream/impl/TipeeestreamEventInputImpl.scala

@@ -19,10 +19,14 @@ class TipeeestreamEventInputImpl extends EventInputImpl[TipeeestreamEvent, Tipee
   private val DATE_FORMATTER = new DateTimeFormatterBuilder()
     .parseCaseInsensitive().append(DateTimeFormatter.ISO_LOCAL_DATE_TIME).appendOffset("+HHMM", "Z").toFormatter
 
+  private val onFollowFn = onFollow _
+  private val onSubscriptionFn = onSubscription _
+  private val onDonationFn = onDonation _
+
   override def start(): Boolean = {
-    sourceConnector.get.addFollowEventListener(onFollow)
-    sourceConnector.get.addSubscriptionEventListener(onSubscription)
-    sourceConnector.get.addDonationEventListener(onDonation)
+    sourceConnector.get.addFollowEventListener(onFollowFn)
+    sourceConnector.get.addSubscriptionEventListener(onSubscriptionFn)
+    sourceConnector.get.addDonationEventListener(onDonationFn)
     true
   }
 
@@ -83,5 +87,10 @@ class TipeeestreamEventInputImpl extends EventInputImpl[TipeeestreamEvent, Tipee
     }
   }
 
-  override def stop(): Boolean = true
+  override def stop(): Boolean = {
+    sourceConnector.get.removeFollowEventListener(onFollowFn)
+    sourceConnector.get.removeSubscriptionEventListener(onSubscriptionFn)
+    sourceConnector.get.removeDonationEventListener(onDonationFn)
+    true
+  }
 }

src/main/scala/org/codeoverflow/chatoverflow/requirement/service/twitch/chat/TwitchChatConnector.scala

@@ -31,6 +31,14 @@ class TwitchChatConnector(override val sourceIdentifier: String) extends Connect
     twitchChatListener.addUnknownEventListener(listener)
   }
 
+  def removeMessageEventListener(listener: MessageEvent => Unit): Unit = {
+    twitchChatListener.removeMessageEventListener(listener)
+  }
+
+  def removeUnknownEventListener(listener: UnknownEvent => Unit): Unit = {
+    twitchChatListener.removeUnknownEventListener(listener)
+  }
+
   def joinChannel(channel: String): Unit = {
     bot.send().joinChannel(channel)
     channels += channel

src/main/scala/org/codeoverflow/chatoverflow/requirement/service/twitch/chat/TwitchChatListener.scala

@@ -29,4 +29,12 @@ class TwitchChatListener extends ListenerAdapter {
     unknownEventListener += listener
   }
 
+  def removeMessageEventListener(listener: MessageEvent => Unit): Unit = {
+    messageEventListener -= listener
+  }
+
+  def removeUnknownEventListener(listener: UnknownEvent => Unit): Unit = {
+    unknownEventListener -= listener
+  }
+
 }

src/main/scala/org/codeoverflow/chatoverflow/requirement/service/twitch/chat/impl/TwitchChatInputImpl.scala

@@ -9,7 +9,7 @@ import org.codeoverflow.chatoverflow.api.io.dto.chat.{ChatEmoticon, TextChannel}
 import org.codeoverflow.chatoverflow.api.io.event.chat.twitch.{TwitchChatMessageReceiveEvent, TwitchEvent, TwitchPrivateChatMessageReceiveEvent}
 import org.codeoverflow.chatoverflow.api.io.input.chat._
 import org.codeoverflow.chatoverflow.registry.Impl
-import org.codeoverflow.chatoverflow.requirement.impl.{EventInputImpl, InputImpl}
+import org.codeoverflow.chatoverflow.requirement.impl.EventInputImpl
 import org.codeoverflow.chatoverflow.requirement.service.twitch.chat
 import org.codeoverflow.chatoverflow.requirement.service.twitch.chat.TwitchChatConnector
 import org.pircbotx.hooks.events.{MessageEvent, UnknownEvent}
@@ -31,9 +31,12 @@ class TwitchChatInputImpl extends EventInputImpl[TwitchEvent, chat.TwitchChatCon
 
   private var currentChannel: Option[String] = None
 
+  private val onMessageFn = onMessage _
+  private val onUnknownFn = onUnknown _
+
   override def start(): Boolean = {
-    sourceConnector.get.addMessageEventListener(onMessage)
-    sourceConnector.get.addUnknownEventListener(onUnknown)
+    sourceConnector.get.addMessageEventListener(onMessageFn)
+    sourceConnector.get.addUnknownEventListener(onUnknownFn)
     true
   }
 
@@ -105,5 +108,9 @@ class TwitchChatInputImpl extends EventInputImpl[TwitchEvent, chat.TwitchChatCon
     *
     * @return true if stopping was successful
     */
-  override def stop(): Boolean = true
+  override def stop(): Boolean = {
+    sourceConnector.get.removeMessageEventListener(onMessageFn)
+    sourceConnector.get.removeUnknownEventListener(onUnknownFn)
+    true
+  }
 }