Cherry picked ClassLibrary Rework from Andre.

Author: Renxiem

build.sbt

@@ -47,6 +47,18 @@ libraryDependencies += "org.pircbotx" % "pircbotx" % "2.1"
 // Reflections API for annotation indexing
 libraryDependencies += "org.reflections" % "reflections" % "0.9.11"
 
+// Akka Actors
+libraryDependencies ++= Seq(
+  "com.typesafe.akka" %% "akka-actor" % "2.5.18",
+  //"com.typesafe.akka" %% "akka-testkit" % "2.5.18" % Test
+)
+
+// Akka Actors
+libraryDependencies ++= Seq(
+  "com.typesafe.akka" %% "akka-actor" % "2.5.18",
+  //"com.typesafe.akka" %% "akka-testkit" % "2.5.18" % Test
+)
+
 
 // ---------------------------------------------------------------------------------------------------------------------
 // PLUGIN FRAMEWORK DEFINITIONS

src/main/scala/org/codeoverflow/chatoverflow/framework/HttpClientActor.scala

@@ -0,0 +1,12 @@
+package org.codeoverflow.chatoverflow.framework
+
+import org.apache.http.client.methods.HttpGet
+import org.apache.http.impl.client.HttpClientBuilder
+
+class HttpClientActor extends Actor {
+  private val client = HttpClientBuilder.create.build
+
+  override def receive: Receive = {
+    case httpGet: HttpGet => sender ! client.execute(httpGet).getEntity
+  }
+}

src/main/scala/org/codeoverflow/chatoverflow/framework/StringMappingActor.scala

@@ -0,0 +1,10 @@
+package org.codeoverflow.chatoverflow.framework
+
+case class Mapping(mapFunc: (String => Any), content: String)
+
+class StringMappingActor extends Actor {
+
+  override def receive: Receive = {
+    case m: Mapping => sender ! m.mapFunc(m.content)
+  }
+}

src/main/scala/org/codeoverflow/chatoverflow/service/twitch/TwitchAPIConnector.scala

@@ -0,0 +1,104 @@
+package org.codeoverflow.chatoverflow.service.twitch
+
+import org.apache.http.HttpEntity
+import org.apache.http.client.methods.HttpGet
+import org.apache.http.client.utils.URIBuilder
+import org.apache.http.util.EntityUtils
+import org.apache.log4j.Logger
+import org.codeoverflow.chatoverflow.configuration.Credentials
+import org.codeoverflow.chatoverflow.framework.HttpClientActor
+
+import scala.concurrent.Await
+import scala.concurrent.duration._
+
+/**
+  * The twitch api connector
+  *
+  * @param sourceIdentifier the name to the twitch account
+  * @param credentials      the credentials to log into the twitch api
+  */
+class TwitchAPIConnector(override val sourceIdentifier: String, credentials: Credentials) extends Connector(sourceIdentifier, credentials) {
+  private val logger = Logger.getLogger(this.getClass)
+  private val API_FORMAT: String = "application/vnd.twitchtv.v5+json"
+  private val BASE_URL: String = "https://api.twitch.tv/helix/"
+  private val BASE_URL_v5: String = "https://api.twitch.tv/kraken/"
+  private val actorSystem = ActorSystem("TwitchAPIActorSystem")
+  private val actor: ActorRef = actorSystem.actorOf(Props[HttpClientActor])
+  private var clientID = ""
+  private var oauth = ""
+
+  override def getUniqueTypeString: String = this.getClass.getName
+
+  /**
+    * Returns true, if the connector has been already instantiated and is running.
+    */
+  override def isRunning: Boolean = true
+
+  /**
+    * Initializes the connector, e.g. creates a connection with its platform.
+    */
+  override def init(): Unit = {
+    val oauth = credentials.getValue(TwitchAPIConnector.credentialsOauthKey)
+    val clientID = credentials.getValue(TwitchAPIConnector.credentialsClientID)
+
+    if (clientID.isEmpty) {
+      logger warn s"key '${TwitchAPIConnector.credentialsClientID}' not found in credentials for '$sourceIdentifier'."
+    } else {
+      this.clientID = clientID.get
+      if (oauth.isEmpty) {
+        logger warn s"key '${TwitchAPIConnector.credentialsOauthKey}' not found in credentials for '$sourceIdentifier'."
+      } else {
+        this.oauth = oauth.get
+      }
+    }
+  }
+
+  def getSubscriptions(channelID: String, offset: Int = 0, newestFirst: Boolean = true) = {
+    get("channels/" + channelID + "/subscriptions", true, true, Seq(("limit", "100"), ("offset", String.valueOf(offset)), ("direction", if (newestFirst) "desc" else "asc")))
+  }
+
+  def getUser(userLogin: String) = {
+    get("users", false, false, Seq(("login", userLogin)))
+  }
+
+  def get(uri: String, auth: Boolean, oldAPI: Boolean, queryParams: Seq[(String, String)]) = {
+    val httpGet = if (auth) getUrlAuth(uri, oldAPI) else getURL(uri, oldAPI)
+    val urlBuilder = new URIBuilder(httpGet.getURI)
+    queryParams.foreach(param => urlBuilder.addParameter(param._1, param._2))
+    httpGet.setURI(urlBuilder.build())
+    implicit val timeout: Timeout = Timeout(5 seconds)
+    val entity = Await.result(actor ? httpGet, timeout.duration).asInstanceOf[HttpEntity]
+    if (entity != null) {
+      EntityUtils.toString(entity, "UTF-8");
+    }
+    else ""
+  }
+
+  def getUrlAuth(uri: String, oldAPI: Boolean) = {
+    val get = getURL(uri, oldAPI)
+    get.setHeader("Authorization", oauth)
+    get
+  }
+
+  def getURL(uri: String, oldAPI: Boolean) = {
+    val baseUrl = if (oldAPI) BASE_URL_v5 else BASE_URL
+    new HttpGet(baseUrl + uri) {
+      setHeader("Accept", API_FORMAT)
+      setHeader("Client-ID", clientID)
+    }
+  }
+
+  def getFollowers(userID: String) = {
+    get("users/follows", false, false, Seq(("to_id", userID)))
+  }
+
+  /**
+    * Shuts down the connector, closes its platform connection.
+    */
+  override def shutdown(): Unit = ???
+}
+
+object TwitchAPIConnector {
+  val credentialsOauthKey = "oauth"
+  val credentialsClientID = "clientid"
+}

src/main/scala/org/codeoverflow/chatoverflow/service/twitch/stat/impl/TwitchStatImpl.scala

@@ -0,0 +1,51 @@
+package org.codeoverflow.chatoverflow.service.twitch.stat.impl
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import org.codeoverflow.chatoverflow.framework.{Mapping, StringMappingActor}
+import org.codeoverflow.chatoverflow.service.twitch.TwitchAPIConnector
+
+import scala.concurrent.Await
+import scala.concurrent.duration._
+
+case class UserResult(data: Seq[UserEntity])
+
+case class UserEntity(id: String, login: String, display_name: String, `type`: String, broadcaster_type: String, description: String, profile_image_url: String, offline_image_url: String, view_count: Int)
+
+class TwitchStatInputImpl extends Connection[TwitchAPIConnector] with TwitchStatInput {
+  private val actorSystem = ActorSystem("TwitchAPIActorSystem")
+  private val actor: ActorRef = actorSystem.actorOf(Props[StringMappingActor])
+  implicit val timeout: Timeout = Timeout(5 seconds)
+
+  override def init(): Unit = {
+    sourceConnector.init()
+  }
+
+  override def getFollowers(userName: String): java.util.List[User] = {
+    val userID = getUser(userName).getId
+    val response = sourceConnector.getFollowers(userID)
+    println(response)
+    null
+  }
+
+  override def getUser(userName: String): User = {
+    val response = sourceConnector.getUser(userName)
+    println(response)
+    val result = Await.result(actor ? Mapping(map[UserResult], response), timeout.duration).asInstanceOf[UserResult]
+    if (result.data.nonEmpty) {
+      val user = result.data.head
+      new User(user.id, user.display_name, user.description, user.profile_image_url, user.view_count)
+    }
+    else null
+  }
+
+  def map[T: Manifest](content: String): Any = {
+    val mapper = new ObjectMapper() with ScalaObjectMapper
+    mapper.registerModule(DefaultScalaModule)
+    mapper.readValue[T](content)
+  }
+
+  override def getSubscribers(userName: String): String = {
+    val userID = getUser(userName).getId
+    sourceConnector.getSubscriptions(userID)
+  }
+}