Initial commit

.gitignore

@@ -0,0 +1,4 @@
+target/
+.idea/
+src/main/resources
+src/main/scala/example

README.md

@@ -0,0 +1,57 @@
+# Twitch IRC
+Twitch irc implementation on scala using akka actors. Twitch server API described [here](https://dev.twitch.tv/docs/irc)
+### Usage
+Twitch bot is an actor which sends messages about all events to its listeners.
+### Messages that can be sent to TwitchIRCActor
+* ```Join(channelName: String)```
+* ```Part(channelName: String)```
+* ```SendMessage(channelName: String, message: String)```
+* ```SendWhisper(userName: String, message: String)```
+* ```AddListener(listener: ActorRef)```
+* ```DeleteListener(listener: ActorRef)```
+### Events
+if ```tags``` is disabled then ```tags``` are empty
+
+success/fail log in events:
+* ```ConnectionFailed(err: String)```
+* ```Connected```
+
+event after logging in(contains information about user in ```tags```)
+* ```GlobalUserState(tags: Map[String, String])```
+
+message events:
+* ```IncomingMessage(tags: Map[String, String], user: String, channel: String, message: String)```
+* ```IncomingWhisper(tags: Map[String, String], user: String, message: String)```
+
+sent only if ```membership``` is enabled:
+* ```UserJoinedChannel(user: String, channel: String)```
+* ```UserLeftChannel(user: String, channel: String)```
+* ```UserGainModeMessage(user: String, channel: String)```
+* ```UserLostModeMessage(user: String, channel: String)```
+* ```ChannelUserList(channel: String, users: Seq[String])```
+
+sent only if ```commands``` is enabled:
+* ```UserBan(tags: Map[String, String], channel: String, user: String)```
+* ```HostStart(channel: String, hostChannel: String, viewers: Option[Int])```
+* ```HostStop(channel: String, viewers: Option[Int])```
+* ```Notice(tags: Map[String, String], channel: String, message: String)```
+* ```RoomState(tags: Map[String, String], channel: String)```
+* ```UserNotice(tags: Map[String, String], channel: String, message: String)```
+* ```UserState(tags: Map[String, String], channel: String)```
+
+### Creating example
+Code inside actor:
+```
+import bot.twitchirc.TwitchIRCActor
+import bot.twitchirc.TwitchIRCActor.{AddListener, TwitchIRCProps}
+
+val twitchIRCActor = context.actorOf(TwitchIRCActor.props(nick, oauth,
+    props = TwitchIRCProps(membership = true, tags = true, commands = true)), "TwitchIRCActor")
+  twitchIRCActor ! AddListener(self)
+```
+```nick``` - twitch login
+```oauth``` - can be obtained [here](https://twitchapps.com/tmi/)
+### TwitchIRCProps - case class with properties:
+* membership - server send information about users join/left. Described [here](https://dev.twitch.tv/docs/irc#twitch-irc-capability-membership).
+* tags - additional information about users/events in messages. Described [here](https://dev.twitch.tv/docs/irc#twitch-irc-capability-tags).
+* commands - adds additional commands. Described [here](https://dev.twitch.tv/docs/irc#twitch-irc-capability-commands).

build.sbt

@@ -0,0 +1,10 @@
+name := "TwitchIRC"
+
+version := "0.1"
+
+scalaVersion := "2.12.5"
+
+libraryDependencies ++= Seq(
+  "com.typesafe.akka" %% "akka-actor" % "2.5.11",
+  "com.typesafe.akka" %% "akka-testkit" % "2.5.11" % Test
+)

project/build.properties

@@ -0,0 +1 @@
+sbt.version = 1.1.2

src/main/scala/bot/twitchirc/TCPClientActor.scala

@@ -0,0 +1,79 @@
+package bot.twitchirc
+
+import java.net.InetSocketAddress
+
+import akka.actor.{Actor, ActorLogging, Props}
+import akka.io.{IO, Tcp}
+import akka.util.ByteString
+import bot.twitchirc.TCPClientActor.{CloseConnection, DataReceived, SendData, WriteFailed, _}
+
+/**
+  * Created by amir.
+  */
+class TCPClientActor(remote: InetSocketAddress) extends Actor with ActorLogging {
+
+  import Tcp._
+  import context.system
+
+  log.info(s"trying to connect ${remote.getHostName}:${remote.getPort} ")
+  private val manager = IO(Tcp)
+  manager ! Connect(remote)
+
+  private var buffer: String = ""
+
+  override def postStop(): Unit = {
+    manager ! Close
+  }
+
+  private val delimiter = List(13, 10).map(_.toChar).mkString
+
+  def receive: Receive = {
+    case CommandFailed(_: Connect) ⇒
+      context.parent ! ConnectionFailed("can't connect to server")
+      context stop self
+    case Connected(_, _) ⇒
+      context.parent ! TCPClientActor.Connected
+      val connection = sender()
+      connection ! Register(self)
+      context become {
+        case SendData(data) ⇒
+          //log.info(s"sending: $data")
+          connection ! Write(ByteString(data + "\n"))
+        case CloseConnection ⇒
+          connection ! Close
+          context stop self
+
+        case CommandFailed(w: Write) ⇒
+          // O/S buffer was full
+          context.parent ! WriteFailed(w.data.toString())
+        case Received(data) ⇒
+          val split = (buffer ++ data.utf8String).split(delimiter, -1)
+          buffer = split.last
+          split.init foreach { line =>
+            //log.info(s"received: $line")
+            context.parent ! DataReceived(line)
+          }
+      }
+  }
+}
+
+object TCPClientActor {
+  def props(remote: InetSocketAddress) =
+    Props(classOf[TCPClientActor], remote)
+
+  //send result of connecting to parent
+  case object Connected
+
+  case class ConnectionFailed(message: String)
+
+  //messages sent to listener on events
+  case class DataReceived(data: String)
+
+  case class WriteFailed(data: String)
+
+  //messages that can be sent
+  case class SendData(data: String)
+
+  case object CloseConnection
+
+}

src/main/scala/bot/twitchirc/TwitchIRCActor.scala

@@ -0,0 +1,198 @@
+package bot.twitchirc
+
+import java.net.InetSocketAddress
+
+import akka.actor.{Actor, ActorLogging, ActorRef, Cancellable, Props}
+import bot.twitchirc.TCPClientActor.{Connected, ConnectionFailed, DataReceived, SendData}
+import bot.twitchirc.TwitchIRCActor._
+import bot.twitchirc.messages.Message._
+import bot.twitchirc.messages.MessageParser
+
+import scala.concurrent.duration._
+
+/**
+  * Created by amir.
+  */
+class TwitchIRCActor(address: InetSocketAddress, nick: String, oauth: String, props: TwitchIRCProps) extends Actor with ActorLogging {
+
+  import context.dispatcher
+
+  assert(nick.forall(!_.isUpper))
+
+  private val respondTimeout = 30 seconds
+
+  private val successfulConnectionMessage = s":tmi.twitch.tv 376 $nick :>"
+  private val successfulMembership = ":tmi.twitch.tv CAP * ACK :twitch.tv/membership"
+  private val successfulTags = ":tmi.twitch.tv CAP * ACK :twitch.tv/tags"
+  private val successfulCommands = ":tmi.twitch.tv CAP * ACK :twitch.tv/commands"
+
+  private val tcpClientActor = context.actorOf(TCPClientActor.props(address), "TCPClient")
+  private var timeoutTerminator: Cancellable = _
+  private var membershipTimeout: Cancellable = _
+  private var tagsTimeout: Cancellable = _
+  private var commandsTimeout: Cancellable = _
+  private val messageParser = new MessageParser(nick)
+  private var listeners: Set[ActorRef] = Set.empty
+
+  private var sentMessages: Map[(String, String), Cancellable] = Map.empty
+  private var joins: Map[String, Cancellable] = Map.empty
+  private var parts: Map[String, Cancellable] = Map.empty
+  private var channelsToUsers: Map[String, Seq[String]] = Map.empty
+
+  override def preStart(): Unit = {
+    log.info("starting TwitchIrc")
+  }
+
+  override def receive: Receive = {
+    case err@ConnectionFailed(_) =>
+      log.error("Connection to twitch server failed. Shutting down.")
+      listeners.foreach(_ ! err)
+      context stop self
+    case Connected =>
+      log.info("Successfully connected to twitch server")
+      if (props.membership) {
+        tcpClientActor ! SendData("CAP REQ :twitch.tv/membership")
+        membershipTimeout = context.system.scheduler.scheduleOnce(respondTimeout)({
+          log.error("Twitch server doesn't respond for membership capability request")
+        })
+      }
+      if (props.tags) {
+        tcpClientActor ! SendData("CAP REQ :twitch.tv/tags")
+        tagsTimeout = context.system.scheduler.scheduleOnce(respondTimeout)({
+          log.error("Twitch server doesn't respond for tags capability request")
+        })
+      }
+      if (props.commands) {
+        tcpClientActor ! SendData("CAP REQ :twitch.tv/commands")
+        commandsTimeout = context.system.scheduler.scheduleOnce(respondTimeout)({
+          log.error("Twitch server doesn't respond for commands capability request")
+        })
+      }
+      //try to login
+      tcpClientActor ! SendData(s"PASS $oauth")
+      tcpClientActor ! SendData(s"NICK $nick")
+      timeoutTerminator = context.system.scheduler.scheduleOnce(respondTimeout)({
+        log.error("Connected to server, but can't log in")
+        listeners.foreach(_ ! ConnectionFailed("Connected to server, but can't log in"))
+        context stop self
+      })
+    case DataReceived(`successfulMembership`) if props.membership =>
+      membershipTimeout.cancel()
+      log.info("membership capability included")
+    case DataReceived(`successfulTags`) if props.tags =>
+      tagsTimeout.cancel()
+      log.info("tags capability included")
+    case DataReceived(`successfulCommands`) if props.commands =>
+      commandsTimeout.cancel()
+      log.info("commands capability included")
+    case DataReceived(`successfulConnectionMessage`) =>
+      timeoutTerminator.cancel()
+      log.info("Successfully logged in")
+      self ! Join("#" + nick)
+      context become commandParser
+      listeners.foreach(_ ! Connected)
+    case AddListener(listener: ActorRef) => listeners += listener
+    case DeleteListener(listener: ActorRef) => listeners -= listener
+  }
+
+  private def commandParser: Receive = {
+    case DataReceived(line) =>
+      messageParser.parseLine(line) match {
+        case Ping =>
+          tcpClientActor ! SendData("PONG :tmi.twitch.tv")
+        case JoinConfirmation(channel) =>
+          joins.get(channel).foreach(_.cancel())
+          joins -= channel
+        case PartConfirmation(channel) =>
+          parts.get(channel).foreach(_.cancel())
+          parts -= channel
+        case MessageDeliverConfirmation(channel, message) =>
+          sentMessages.get((channel, message)).foreach(_.cancel())
+          sentMessages -= ((channel, message))
+
+        case ChannelUserList(channel, users) =>
+          channelsToUsers += channel -> (users ++ channelsToUsers.getOrElse(channel, Seq()))
+        case EndOfUserList(channel) =>
+          listeners.foreach(_ ! ChannelUserList(channel, channelsToUsers.getOrElse(channel, Seq())))
+          channelsToUsers -= channel
+
+        case msg@(UserJoinedChannel(_, _)
+                  | UserLeftChannel(_, _)
+                  | UserGainModeMessage(_, _)
+                  | UserLostModeMessage(_, _)
+                  | IncomingMessage(_, _, _, _)
+                  | UserBan(_, _, _)
+                  | GlobalUserState(_)
+                  | RoomState(_, _)
+                  | UserNotice(_, _, _)
+                  | UserState(_, _)
+                  | HostStart(_, _, _)
+                  | HostStop(_, _)
+                  | Notice(_, _, _)) => listeners.foreach(_ ! msg)
+        case msg@UnknownMessage(_) => listeners.foreach(_ ! msg)
+      }
+    case Join(name) =>
+      val channelName = name.toLowerCase
+      tcpClientActor ! SendData(s"JOIN $channelName")
+      joins += channelName -> context.system.scheduler.scheduleOnce(respondTimeout)({
+        log.error(s"Twitch server doesn't respond JOIN $channelName request")
+        joins -= channelName
+      })
+    case Part(name) =>
+      val channelName = name.toLowerCase
+      tcpClientActor ! SendData(s"PART $channelName")
+      parts += channelName -> context.system.scheduler.scheduleOnce(respondTimeout)({
+        log.error(s"Twitch server doesn't respond PART $channelName request")
+        parts -= channelName
+      })
+    case SendMessage(name, message) =>
+      val channelName = name.toLowerCase
+      tcpClientActor ! SendData(s"PRIVMSG $channelName :$message")
+      sentMessages += (channelName, message) -> context.system.scheduler.scheduleOnce(respondTimeout)({
+        log.error(s"Twitch server doesn't respond PRIVMSG $channelName, $message request")
+        parts -= (channelName, message)
+      })
+    case SendWhisper(name, message) =>
+      val userName = name.toLowerCase
+      tcpClientActor ! SendData(s"PRIVMSG #$nick :/w $userName $message")
+    case AddListener(listener: ActorRef) => listeners += listener
+    case DeleteListener(listener: ActorRef) => listeners -= listener
+  }
+
+}
+
+object TwitchIRCActor {
+
+  /**
+    *
+    * @param membership Adds membership state event data
+    * @param tags       Adds IRC V3 message tags to several commands, if enabled with the commands capability.
+    * @param commands   Enables several Twitch-specific commands
+    * @param botType    throttling parameter
+    */
+  case class TwitchIRCProps(membership: Boolean = true, tags: Boolean = true, commands: Boolean = true, botType: BotType = KnownBot)
+
+  sealed trait BotType
+  case object KnownBot extends BotType
+
+  private val defaultAddress = new InetSocketAddress("irc.chat.twitch.tv", 6667)
+
+  private val defaultProps = TwitchIRCProps()
+
+  def props(nick: String, oauth: String, address: InetSocketAddress = defaultAddress, props: TwitchIRCProps = defaultProps): Props =
+    Props(new TwitchIRCActor(address, nick.toLowerCase, oauth, props))
+
+  case class Join(channelName: String)
+
+  case class Part(channelName: String)
+
+  case class SendMessage(channelName: String, message: String)
+
+  case class SendWhisper(userName: String, message: String)
+
+  case class AddListener(listener: ActorRef)
+
+  case class DeleteListener(listener: ActorRef)
+
+
+}

src/main/scala/bot/twitchirc/messages/Message.scala

@@ -0,0 +1,37 @@
+package bot.twitchirc.messages
+
+/**
+  * Created by amir.
+  */
+sealed trait Message
+
+object Message {
+
+  private[twitchirc] case object Ping extends Message
+
+  case class UnknownMessage(message: String) extends Message
+
+  private[twitchirc] case class JoinConfirmation(channel: String) extends Message
+  private[twitchirc] case class PartConfirmation(channel: String) extends Message
+  private[twitchirc] case class MessageDeliverConfirmation(channel: String, message: String) extends Message
+
+  case class UserJoinedChannel(user: String, channel: String) extends Message
+  case class UserLeftChannel(user: String, channel: String) extends Message
+  case class UserGainModeMessage(user: String, channel: String) extends Message
+  case class UserLostModeMessage(user: String, channel: String) extends Message
+  case class ChannelUserList(channel: String, users: Seq[String]) extends Message
+  private[twitchirc] case class EndOfUserList(channel: String) extends Message
+
+  case class IncomingMessage(tags: Map[String, String], user: String, channel: String, message: String) extends Message
+  case class IncomingWhisper(tags: Map[String, String], user: String, message: String) extends Message
+  case class UserBan(tags: Map[String, String], channel: String, user: String) extends Message
+  case class GlobalUserState(tags: Map[String, String]) extends Message
+  case class RoomState(tags: Map[String, String], channel: String) extends Message
+  case class UserNotice(tags: Map[String, String], channel: String, message: String) extends Message
+  case class UserState(tags: Map[String, String], channel: String) extends Message
+
+  case class HostStart(channel: String, hostChannel: String, viewers: Option[Int]) extends Message
+  case class HostStop(channel: String, viewers: Option[Int]) extends Message
+  case class Notice(tags: Map[String, String], channel: String, message: String) extends Message
+
+}