Also use EventManager for Tipeeestream

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

@@ -4,24 +4,26 @@ import java.util.Calendar
 
 import io.socket.client.{IO, Socket}
 import org.codeoverflow.chatoverflow.WithLogger
-import org.codeoverflow.chatoverflow.connector.Connector
+import org.codeoverflow.chatoverflow.connector.EventConnector
 import org.json.JSONObject
 
 /**
   * The tipeeestream connector connects to the socket.io service to work with incoming events.
   *
   * @param sourceIdentifier the name of the tipeeestream account
   */
-class TipeeestreamConnector(override val sourceIdentifier: String) extends Connector(sourceIdentifier) with WithLogger {
+class TipeeestreamConnector(override val sourceIdentifier: String) extends EventConnector(sourceIdentifier) with WithLogger {
   private val TIMEOUT = 10000
   private val SOCKET_URL = "https://sso-cf.tipeeestream.com"
   private val tipeeeStreamListener = new TipeeestreamListener
   override protected var requiredCredentialKeys: List[String] = List("apiKey", "username")
   override protected var optionalCredentialKeys: List[String] = List()
   private var socket: Option[Socket] = None
 
+  tipeeeStreamListener.registerEventHandler((event, ct) => call(event)(ct, ct))
+
   override def start(): Boolean = {
-    //RestAPI doesn't need stratup methods
+    // RestAPI doesn't need startup methods
     startSocket()
   }
 
@@ -80,20 +82,15 @@ class TipeeestreamConnector(override val sourceIdentifier: String) extends Conne
     obj
   }
 
-  def addSubscriptionEventListener(listener: JSONObject => Unit): Unit = tipeeeStreamListener.addSubscriptionEventListener(listener)
-
-  def addDonationEventListener(listener: JSONObject => Unit): Unit = tipeeeStreamListener.addDonationEventListener(listener)
-
-  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
   }
 }
+
+object TipeeestreamConnector {
+  private[tipeeestream] sealed class TipeeestreamEventJSON(json: JSONObject)
+  private[tipeeestream] case class SubscriptionEventJSON(json: JSONObject) extends TipeeestreamEventJSON(json)
+  private[tipeeestream] case class DonationEventJSON(json: JSONObject) extends TipeeestreamEventJSON(json)
+  private[tipeeestream] case class FollowEventJSON(json: JSONObject) extends TipeeestreamEventJSON(json)
+}

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

@@ -1,52 +1,24 @@
 package org.codeoverflow.chatoverflow.requirement.service.tipeeestream
 
-
+import org.codeoverflow.chatoverflow.requirement.impl.EventManager
+import org.codeoverflow.chatoverflow.requirement.service.tipeeestream.TipeeestreamConnector._
 import org.json.JSONObject
 
-import scala.collection.mutable.ListBuffer
-
-class TipeeestreamListener {
-
-  private val subscriptionEventListeners = ListBuffer[JSONObject => Unit]()
-  private val donationEventListeners = ListBuffer[JSONObject => Unit]()
-  private val followEventListeners = ListBuffer[JSONObject => Unit]()
-
-  def addSubscriptionEventListener(listener: JSONObject => Unit): Unit = {
-    subscriptionEventListeners += listener
-  }
-
-  def addDonationEventListener(listener: JSONObject => Unit): Unit = {
-    donationEventListeners += listener
-  }
-
-  def addFollowEventListener(listener: JSONObject => Unit): Unit = {
-    followEventListeners += listener
-  }
+class TipeeestreamListener extends EventManager {
 
-  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 = {
+  def onSocketEvent(objects: Array[AnyRef]): Unit = {
     val json: JSONObject = objects(0).asInstanceOf[JSONObject]
     val event: JSONObject = json.getJSONObject("event")
     val eventType: String = event.getString("type")
-    eventType match {
-      case "subscription" =>
-        subscriptionEventListeners.foreach(_(event))
-      case "donation" =>
-        donationEventListeners.foreach(_(event))
-      case "follow" =>
-        followEventListeners.foreach(_(event))
-      case _ =>
-    }
+
+    val eventOption: Option[TipeeestreamEventJSON] = Option(eventType match {
+      case "subscription" => SubscriptionEventJSON(event)
+      case "donation" => DonationEventJSON(event)
+      case "follow" => FollowEventJSON(event)
+      case _ => null // gets None if converted to an option
+    })
+
+    if (eventOption.isDefined)
+      call(eventOption.get) // send event to connector
   }
 }

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

@@ -19,14 +19,10 @@ 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(onFollowFn)
-    sourceConnector.get.addSubscriptionEventListener(onSubscriptionFn)
-    sourceConnector.get.addDonationEventListener(onDonationFn)
+    sourceConnector.get.registerEventHandler(onFollow _)
+    sourceConnector.get.registerEventHandler(onSubscription _)
+    sourceConnector.get.registerEventHandler(onDonation _)
     true
   }
 
@@ -88,9 +84,7 @@ class TipeeestreamEventInputImpl extends EventInputImpl[TipeeestreamEvent, Tipee
   }
 
   override def stop(): Boolean = {
-    sourceConnector.get.removeFollowEventListener(onFollowFn)
-    sourceConnector.get.removeSubscriptionEventListener(onSubscriptionFn)
-    sourceConnector.get.removeDonationEventListener(onDonationFn)
+    sourceConnector.get.unregisterAllEventListeners
     true
   }
 }