<commit_before>package actors

import java.util

import akka.actor.Actor
import akka.actor.ActorLogging
import akka.event.LoggingReceive
import akka.actor.ActorRef
import akka.actor.Terminated
import play.libs.Akka
import akka.actor.Props
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global

class ChannelActor extends Actor with ActorLogging {
  var onlineUsers = Map[ActorRef, String]()

  var posts = List[Message]()

  def receive = LoggingReceive {
    /**
     * Send this message to the each user which registered to this channel
     */
    case m: Message =>
      posts +:= m

      onlineUsers.keys foreach (_ ! m)

    case m: StatusUserTyping => onlineUsers.keys foreach (_ ! m)

    /**
     * Subscribe new user to this channel
     */
    case Subscribe(uid) =>
      onlineUsers += sender -> uid

      //watch the sender actor for Termination so we can deRegister them
      context watch sender

      //send last 300 message for each channel to the new User
      //todo send only users subscribed channels
      sender ! Message("_replyingChannelHistory_STARTED", "mark")
      (posts groupBy (_.channel) flatMap (_._2 take 300) toList).reverse foreach (sender ! _)
      sender ! Message("_replyingChannelHistory_FINISHED", "mark")

      //      sender ! Message("Cowboy Bebop (BOT)", "Gotta Knock a Little Harder!")
      //      sender ! Message("Shebang", s"Hello welcome back #!")
      onlineUsers.keys foreach (_ ! Message("_userStatusChanged_ONLINE", uid))


    /**
     * broadcast number of online users
     */
    case BroadcastStatus =>
      onlineUsers.keys foreach (_ ! Message("_numberOfOnlineUsers", s"${onlineUsers.size}"))


    /**
     * deRegister the user from this channel on user (UserActor) termination.
     */
    case Terminated(user) => onlineUsers.get(user).map { uid =>
      onlineUsers -= user
      onlineUsers foreach { case (userActor, _) => userActor ! Message("_userStatusChanged_OFFLINE", uid)}
    }


  }

  /**
   * Every 30 seconds broadcast number of online users
   */
  override def preStart() = {
    context.system.scheduler.schedule(3 seconds, 60 seconds, self, BroadcastStatus)
  }

  ChannelActor()
<commit_msg>onliner user initial broadcast fix<commit_after>package actors

import java.util

import akka.actor.Actor
import akka.actor.ActorLogging
import akka.event.LoggingReceive
import akka.actor.ActorRef
import akka.actor.Terminated
import play.libs.Akka
import akka.actor.Props
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global

class ChannelActor extends Actor with ActorLogging {
  var onlineUsers = Map[ActorRef, String]()

  var posts = List[Message]()

  def receive = LoggingReceive {
    /**
     * Send this message to the each user which registered to this channel
     */
    case m: Message =>
      posts +:= m

      onlineUsers.keys foreach (_ ! m)

    case m: StatusUserTyping => onlineUsers.keys foreach (_ ! m)

    /**
     * Subscribe new user to this channel
     */
    case Subscribe(uid) =>
      onlineUsers += sender -> uid

      //watch the sender actor for Termination so we can deRegister them
      context watch sender

      //send last 300 message for each channel to the new User
      //todo send only users subscribed channels
      sender ! Message("_replyingChannelHistory_STARTED", "mark")
      (posts groupBy (_.channel) flatMap (_._2 take 300) toList).reverse foreach (sender ! _)
      sender ! Message("_replyingChannelHistory_FINISHED", "mark")

      //      sender ! Message("Cowboy Bebop (BOT)", "Gotta Knock a Little Harder!")
      //      sender ! Message("Shebang", s"Hello welcome back #!")

      //send status to every online user
      onlineUsers.keys foreach (_ ! Message("_userStatusChanged_ONLINE", uid))
      //send every other online users status to sender
      onlineUsers.values.filterNot(_ == uid) foreach (sender ! Message("_userStatusChanged_ONLINE", _))


    /**
     * broadcast number of online users
     */
    case BroadcastStatus =>
      onlineUsers.keys foreach (_ ! Message("_numberOfOnlineUsers", s"${onlineUsers.size}"))


    /**
     * deRegister the user from this channel on user (UserActor) termination.
     */
    case Terminated(user) => onlineUsers.get(user).map { uid =>
      onlineUsers -= user
      onlineUsers foreach { case (userActor, _) => userActor ! Message("_userStatusChanged_OFFLINE", uid)}
    }


  }

  /**
   * Every 30 seconds broadcast number of online users
   */
  override def preStart() = {
    context.system.scheduler.schedule(3 seconds, 60 seconds, self, BroadcastStatus)
  }

  ChannelActor()

{'fraction_non_alphanumeric': 0.10004275331338179, 'fraction_numerical': 0.005130397605814451, 'mean_word_length': 3.148049645390071, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5190142', 'n_tokens_mistral': 1355, 'n_tokens_neox': 1282, 'n_words': 409}