Merge branch 'develop' of https://github.com/codeoverflow-org/chatoverflow into develop

Author: sebinside

src/main/scala/ScalatraBootstrap.scala

@@ -1,6 +1,7 @@
 import javax.servlet.ServletContext
 import org.codeoverflow.chatoverflow.ui.web.rest.config.ConfigController
 import org.codeoverflow.chatoverflow.ui.web.rest.connector.ConnectorController
+import org.codeoverflow.chatoverflow.ui.web.rest.events.{EventsController, EventsDispatcher}
 import org.codeoverflow.chatoverflow.ui.web.rest.plugin.PluginInstanceController
 import org.codeoverflow.chatoverflow.ui.web.rest.types.TypeController
 import org.codeoverflow.chatoverflow.ui.web.{CodeOverflowSwagger, OpenAPIServlet}
@@ -21,6 +22,9 @@ class ScalatraBootstrap extends LifeCycle {
     context.initParameters("org.scalatra.cors.allowedMethods") = "*"
 
     // Add all servlets and controller
+    val eventsController = new EventsController()
+    EventsDispatcher.init(eventsController)
+    context.mount(eventsController, "/events/*", "events")
     context.mount(new TypeController(), "/types/*", "types")
     context.mount(new ConfigController(), "/config/*", "config")
     context.mount(new PluginInstanceController(), "/instances/*", "instances")

src/main/scala/org/codeoverflow/chatoverflow/connector/actor/JsonActor.scala

@@ -4,6 +4,7 @@ import java.util
 
 import org.codeoverflow.chatoverflow.WithLogger
 import org.codeoverflow.chatoverflow.api.plugin.{PluginLogMessage, PluginManager}
+import org.codeoverflow.chatoverflow.ui.web.rest.events.{EventMessage, EventsDispatcher}
 
 import scala.collection.JavaConverters._
 import scala.collection.mutable.ListBuffer
@@ -23,11 +24,18 @@ class PluginManagerImpl(pluginInstanceName: String, logOutputOnConsole: Boolean)
     * @param message the message to show
     */
   override def log(message: String): Unit = {
-    logMessages += new PluginLogMessage(message)
+    val logMessage = new PluginLogMessage(message)
+    logMessages += logMessage
 
     if (logOutputOnConsole) {
       logger info s"[$pluginInstanceName] $message"
     }
+
+    EventsDispatcher.broadcast("instance", EventMessage("log", Map(
+      ("name", pluginInstanceName),
+      ("message", message),
+      ("timestamp", logMessage.getTimestamp.toString)
+    )))
   }
 
   /**

src/main/scala/org/codeoverflow/chatoverflow/framework/manager/PluginManagerImpl.scala

@@ -8,6 +8,7 @@ import org.codeoverflow.chatoverflow.api.plugin.{Plugin, PluginManager}
 import org.codeoverflow.chatoverflow.framework.PluginCompatibilityState.PluginCompatibilityState
 import org.codeoverflow.chatoverflow.framework.manager.{PluginManagerImpl, PluginManagerStub}
 import org.codeoverflow.chatoverflow.framework.{PluginCompatibilityState, PluginType}
+import org.codeoverflow.chatoverflow.ui.web.rest.events.{EventMessage, EventsDispatcher}
 
 /**
   * A plugin instance holds all the general information of the plugin type and specific information of
@@ -149,6 +150,8 @@ class PluginInstance(val instanceName: String, pluginType: PluginType) extends W
             logger info s"Starting plugin '$instanceName' in new thread!"
             try {
               instanceThread = new Thread(() => {
+                EventsDispatcher.broadcast("instance", EventMessage("start", Map(("name", instanceName))))
+
                 try {
 
                   // Execute plugin setup
@@ -191,6 +194,7 @@ class PluginInstance(val instanceName: String, pluginType: PluginType) extends W
                     requirement.asInstanceOf[Requirement[Output]].get().shutdown()
                   })
 
+                  EventsDispatcher.broadcast("instance", EventMessage("stop", Map(("name", instanceName))))
                 }
               })
               instanceThread.start()

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

@@ -4,7 +4,7 @@ import org.codeoverflow.chatoverflow.WithLogger
 import org.codeoverflow.chatoverflow.connector.Connector
 import org.reflections.Reflections
 import org.reflections.scanners.{SubTypesScanner, TypeAnnotationsScanner}
-import org.reflections.util.{ClasspathHelper, ConfigurationBuilder}
+import org.reflections.util.{ClasspathHelper, ConfigurationBuilder, FilterBuilder}
 
 import scala.collection.mutable
 import scala.collection.mutable.ListBuffer
@@ -55,6 +55,8 @@ class TypeRegistry(requirementPackage: String) extends WithLogger {
     // Use reflection magic to get all impl-annotated classes
     val reflections: Reflections = new Reflections(new ConfigurationBuilder()
       .setUrls(ClasspathHelper.forPackage(requirementPackage))
+      .filterInputsBy(new FilterBuilder()
+        .includePackage(requirementPackage))
       .setScanners(new SubTypesScanner(), new TypeAnnotationsScanner()))
     val classes = reflections.getTypesAnnotatedWith(classOf[Impl])
 

src/main/scala/org/codeoverflow/chatoverflow/registry/TypeRegistry.scala

@@ -35,9 +35,9 @@ class TipeeestreamConnector(override val sourceIdentifier: String) extends Conne
     val thread = Thread.currentThread
     socket = Some(IO.socket(SOCKET_URL).connect())
     socket.get.on(Socket.EVENT_CONNECT, (_: Any) => {
-      logger info "Connected to TipeeStream Socket.io"
+      logger info "Connected to TipeeeStream Socket.io"
       socket.get.emit("join-room", AUTH_OBJECT)
-      logger info "emitted credentials to TipeeSetream Socket.io api"
+      logger info "emitted credentials to TipeeeStream Socket.io api"
       socket.get.on("new-event", (objects: Array[AnyRef]) => {
         tipeeeStreamListener.onSocketEvent(objects)
       })

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

@@ -7,15 +7,15 @@ import java.util.Currency
 import org.codeoverflow.chatoverflow.WithLogger
 import org.codeoverflow.chatoverflow.api.io.dto.User
 import org.codeoverflow.chatoverflow.api.io.dto.stat.stream.tipeeestream.{TipeeestreamDonation, TipeeestreamFollow, TipeeestreamProvider, TipeeestreamSubscription}
-import org.codeoverflow.chatoverflow.api.io.event.stream.tipeeestream.{TipeeestramFollowEvent, TipeeestreamDonationEvent, TipeeestreamEvent, TipeeestreamSubscriptionEvent}
+import org.codeoverflow.chatoverflow.api.io.event.stream.tipeeestream.{TipeeestreamFollowEvent, TipeeestreamDonationEvent, TipeeestreamEvent, TipeeestreamSubscriptionEvent}
 import org.codeoverflow.chatoverflow.api.io.input.event.TipeeestreamEventInput
 import org.codeoverflow.chatoverflow.registry.Impl
 import org.codeoverflow.chatoverflow.requirement.impl.EventInputImpl
 import org.codeoverflow.chatoverflow.requirement.service.tipeeestream.TipeeestreamConnector
 import org.json.{JSONException, JSONObject}
 
 @Impl(impl = classOf[TipeeestreamEventInput], connector = classOf[TipeeestreamConnector])
-class TipeestreamEventInputImpl extends EventInputImpl[TipeeestreamEvent, TipeeestreamConnector] with TipeeestreamEventInput with WithLogger {
+class TipeeestreamEventInputImpl extends EventInputImpl[TipeeestreamEvent, TipeeestreamConnector] with TipeeestreamEventInput with WithLogger {
   private val DATE_FORMATTER = new DateTimeFormatterBuilder()
     .parseCaseInsensitive().append(DateTimeFormatter.ISO_LOCAL_DATE_TIME).appendOffset("+HHMM", "Z").toFormatter
 
@@ -72,7 +72,7 @@ class TipeestreamEventInputImpl extends EventInputImpl[TipeeestreamEvent, Tipeee
       val time = OffsetDateTime.parse(event.getString("created_at"), DATE_FORMATTER)
       val provider = TipeeestreamProvider.parse(event.getString("origin"))
       val follow = new TipeeestreamFollow(user, time, provider)
-      call(new TipeeestramFollowEvent(follow))
+      call(new TipeeestreamFollowEvent(follow))
     } catch {
       case e: JSONException =>
         logger warn "Error while parsing follow json:"

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

@@ -0,0 +1,25 @@
+package org.codeoverflow.chatoverflow.requirement.service.twitch.chat
+
+import org.pircbotx.hooks.events.{ConnectAttemptFailedEvent, ConnectEvent, NoticeEvent}
+import org.pircbotx.hooks.{Event, ListenerAdapter}
+
+/**
+  * Handles connection events for the TwitchChatConnector.
+  * Calls the callback function once the bot connected and reports connection errors.
+  * @param fn the callback which will be called once suitable event has been received.
+  *           The first param informs whether the connection could be established successfully
+  *           and the second param includes a error description if something has gone wrong.
+  */
+class TwitchChatConnectListener(fn: (Boolean, String) => Unit) extends ListenerAdapter {
+  override def onEvent(event: Event): Unit = {
+    event match {
+      case _: ConnectEvent => fn(true, "")
+      case e: ConnectAttemptFailedEvent => fn(false, "couldn't connect to irc chat server")
+      case e: NoticeEvent =>
+        if (e.getNotice.contains("authentication failed")) {
+          fn(false, "authentication failed")
+        }
+      case _ =>
+    }
+  }
+}

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

@@ -15,10 +15,12 @@ import scala.collection.mutable.ListBuffer
   */
 class TwitchChatConnector(override val sourceIdentifier: String) extends Connector(sourceIdentifier) with WithLogger {
   private val twitchChatListener = new TwitchChatListener
+  private val connectionListener = new TwitchChatConnectListener(onConnect)
   private val oauthKey = "oauth"
   override protected var requiredCredentialKeys: List[String] = List(oauthKey)
   override protected var optionalCredentialKeys: List[String] = List()
   private var bot: PircBotX = _
+  private var status: Option[(Boolean, String)] = None
   private val channels = ListBuffer[String]()
 
   def addMessageEventListener(listener: MessageEvent => Unit): Unit = {
@@ -63,6 +65,7 @@ class TwitchChatConnector(override val sourceIdentifier: String) extends Connect
         .setName(credentials.get.credentialsIdentifier)
         .setServerPassword(password.getOrElse(""))
         .addListener(twitchChatListener)
+        .addListener(connectionListener)
         .buildConfiguration()
     } else {
       logger error "No credentials set!"
@@ -71,33 +74,47 @@ class TwitchChatConnector(override val sourceIdentifier: String) extends Connect
 
   }
 
+  /**
+    * Gets called by the TwitchChatConnectListener when the bot has connected.
+    * Saves the passed information into the status variable.
+    */
+  private def onConnect(success: Boolean, msg: String): Unit = {
+    status.synchronized {
+      // tell the thread which starts the connector that the status has been reported
+      status.notify()
+      status = Some((success, msg))
+    }
+  }
+
   /**
     * Starts the connector, e.g. creates a connection with its platform.
     */
   override def start(): Boolean = {
     bot = new PircBotX(getConfig)
     startBot()
-    true
   }
 
-  private def startBot(): Unit = {
-
-    var errorCount = 0
-
+  private def startBot(): Boolean = {
     new Thread(() => {
       bot.startBot()
     }).start()
 
-    while (bot.getState != PircBotX.State.CONNECTED && errorCount < 30) {
-      logger info "Waiting while the bot is connecting..."
-      Thread.sleep(100)
-      errorCount += 1
+    logger info "Waiting while the bot is connecting and logging in..."
+    status.synchronized {
+      status.wait(10000)
+    }
+
+    if (status.isEmpty) {
+      logger error "Bot couldn't connect within timeout of 10 seconds."
+      return false
     }
 
-    if (errorCount >= 30) {
-      logger error "Fatal. Unable to start bot."
+    val (success, msg) = status.get
+    if (!success) {
+      logger error s"Bot couldn't connect. Reason: $msg."
     }
 
+    success
   }
 
   /**
@@ -106,6 +123,8 @@ class TwitchChatConnector(override val sourceIdentifier: String) extends Connect
   override def stop(): Boolean = {
     bot.sendIRC().quitServer()
     bot.close()
+    status = None
+    channels.clear()
     true
   }
 }

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

@@ -45,7 +45,8 @@ class TwitchChatInputImpl extends EventInputImpl[TwitchEvent, chat.TwitchChatCon
       val moderator = event.getV3Tags.get("mod") == "1"
       val broadcaster = event.getV3Tags.get("badges").contains("broadcaster/1")
       val turbo = event.getV3Tags.get("badges").contains("turbo/1")
-      val author = new TwitchChatMessageAuthor(event.getUser.getNick, color, broadcaster, moderator, subscriber, turbo)
+      val vip = event.getV3Tags.get("badges").contains("vip/1")
+      val author = new TwitchChatMessageAuthor(event.getUser.getNick, color, broadcaster, moderator, subscriber, turbo, vip)
       val time = OffsetDateTime.ofInstant(Instant.ofEpochMilli(event.getTimestamp), ZoneOffset.UTC)
       val channel = new TextChannel(event.getChannelSource)
       val emoticons = new java.util.ArrayList[ChatEmoticon]()

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

@@ -0,0 +1,3 @@
+package org.codeoverflow.chatoverflow.ui.web.rest.events
+
+case class EventMessage[T](action: String, data: T)

src/main/scala/org/codeoverflow/chatoverflow/ui/web/rest/events/EventMessage.scala

@@ -0,0 +1,84 @@
+package org.codeoverflow.chatoverflow.ui.web.rest.events
+
+import java.io.PrintWriter
+import java.util.concurrent.ConcurrentHashMap
+
+import javax.servlet.AsyncContext
+import javax.servlet.http.HttpServletRequest
+import org.codeoverflow.chatoverflow.ui.web.JsonServlet
+import org.scalatra.servlet.ScalatraAsyncSupport
+import org.scalatra.{BadRequest, Unauthorized}
+import org.scalatra.swagger.Swagger
+
+class EventsController(implicit val swagger: Swagger) extends JsonServlet with ScalatraAsyncSupport with EventsControllerDefinition {
+  private val connectionWriters = new ConcurrentHashMap[AsyncContext, PrintWriter]()
+
+  def broadcast(messageType: String, message: String = null): Unit = {
+    connectionWriters.forEach((_, writer) => {
+      try {
+        sendMessage(writer, messageType, message)
+      } catch {
+        //probably lost or closed connection, remove from the list of connected clients
+        case _: Throwable => connectionWriters.remove(writer)
+      }
+    })
+  }
+
+  def closeConnections(): Unit = {
+    connectionWriters.forEach((_, writer) => {
+      try {
+        sendMessage(writer, "close", null)
+        writer.close()
+      } finally {
+        connectionWriters.remove(writer)
+      }
+    })
+  }
+
+  private def sendMessage(writer: PrintWriter, messageType: String, message: String): Unit = {
+    /*
+      Every message has the following format and ends with two line feeds (\n):
+      event: [name of event]
+      data: [first line]
+      data: [second line]
+      ...
+
+      See also: https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#Examples
+     */
+
+    var msg = "event: " + messageType.replace("\n", "") + "\n"
+    if (message != null)
+      msg += "data: " + message.replace("\n", "\ndata: ") + "\n\n"
+    writer.write(msg)
+    writer.flush()
+  }
+
+  get("/", operation(getEvents)) {
+    val accept = request.getHeader("Accept")
+    if (accept == null || !accept.replace(" ", "").split(",").contains("text/event-stream")) {
+      status = 406
+    } else {
+      authParamRequired {
+        contentType = "text/event-stream"
+
+        val asyncContext = request.startAsync()
+        asyncContext.setTimeout(0)
+
+        val writer = asyncContext.getResponse.getWriter
+        connectionWriters.put(asyncContext, writer)
+      }
+    }
+  }
+
+  private def authParamRequired(func: => Any)(implicit request: HttpServletRequest): Any = {
+    val authKeyKey = "authKey"
+
+    if (!request.parameters.contains(authKeyKey) || request.getParameter(authKeyKey).isEmpty) {
+      BadRequest()
+    } else if (request.getParameter(authKeyKey) != chatOverflow.credentialsService.generateAuthKey()) {
+      Unauthorized()
+    } else {
+      func
+    }
+  }
+}