<commit_before>package xitrum.sockjs

import scala.collection.mutable.ArrayBuffer

import akka.actor.{Actor, ActorRef, ActorSystem, Props, ReceiveTimeout, Terminated}
import scala.concurrent.duration._

import xitrum.{Action, SockJsHandler}
import xitrum.routing.Routes

sealed trait MessageToSession
case object SubscribeByClient                           extends MessageToSession
case class  SendMessagesByClient(messages: Seq[String]) extends MessageToSession
case class  SendMessageByHandler(message: String)       extends MessageToSession
case object CloseByHandler                              extends MessageToSession

sealed trait SubscribeResultToClient
case object SubscribeResultToClientAnotherConnectionStillOpen      extends SubscribeResultToClient
case object SubscribeResultToClientClosed                          extends SubscribeResultToClient
case class  SubscribeResultToClientMessages(messages: Seq[String]) extends SubscribeResultToClient
case object SubscribeResultToClientWaitForMessage                  extends SubscribeResultToClient

sealed trait NotificationToClient
case class  NotificationToClientMessage(message: String) extends NotificationToClient
case object NotificationToClientHeartbeat                extends NotificationToClient
case object NotificationToClientClosed                   extends NotificationToClient

object NonWebSocketSession {
  // The session must time out after 5 seconds of not having a receiving connection
  // http://sockjs.github.com/sockjs-protocol/sockjs-protocol-0.3.3.html#section-46
  private val TIMEOUT_CONNECTION = 5.seconds

  private val TIMEOUT_CONNECTION_MILLIS = TIMEOUT_CONNECTION.toMillis

  // The server must send a heartbeat frame every 25 seconds
  // http://sockjs.github.com/sockjs-protocol/sockjs-protocol-0.3.3.html#section-46
  private val TIMEOUT_HEARTBEAT = 25.seconds
}

/**
 * There should be at most one subscriber:
 * http://sockjs.github.com/sockjs-protocol/sockjs-protocol-0.3.3.html
 *
 * To avoid out of memory, the actor is stopped when there's no subscriber
 * for a long time. Timeout is also used to check if there's no message
 * for subscriber for a long time.
 * See TIMEOUT_CONNECTION and TIMEOUT_HEARTBEAT in NonWebSocketSessions.
 */
class NonWebSocketSession(var clientSubscriber: ActorRef, pathPrefix: String, action: Action) extends Actor {
  import NonWebSocketSession._

  private var sockJsHandler: SockJsHandler = _

  // Messages from handler to client are buffered here
  private val bufferForClientSubscriber = ArrayBuffer[String]()

  // ReceiveTimeout may not occurred if there's frequent Publish, thus we
  // need to manually check if there's no subscriber for a long time.
  // lastSubscribedAt must be Long to avoid Integer overflow, beacuse
  // System.currentTimeMillis() is used.
  private var lastSubscribedAt = 0L

  // Until the timeout occurs, the server must constantly serve
  // the close message
  private var closed = false

  override def preStart() {
    // FIXME

    // sockJsHandler.onClose is called at postStop
    //sockJsHandler = Routes.createSockJsHandler(pathPrefix)
    sockJsHandler.nonWebSocketSessionActorRef = self
    sockJsHandler.onOpen(action)

    lastSubscribedAt = System.currentTimeMillis()

    context.watch(clientSubscriber)  // Unsubscribed when stopped

    // Once set, the receive timeout stays in effect (i.e. continues firing<commit_msg>Fix NPE at NonWebSocketSession's preStart<commit_after>package xitrum.sockjs

import scala.collection.mutable.ArrayBuffer

import akka.actor.{Actor, ActorRef, ActorSystem, Props, ReceiveTimeout, Terminated}
import scala.concurrent.duration._

import xitrum.{Action, SockJsHandler}
import xitrum.routing.Routes

sealed trait MessageToSession
case object SubscribeByClient                           extends MessageToSession
case class  SendMessagesByClient(messages: Seq[String]) extends MessageToSession
case class  SendMessageByHandler(message: String)       extends MessageToSession
case object CloseByHandler                              extends MessageToSession

sealed trait SubscribeResultToClient
case object SubscribeResultToClientAnotherConnectionStillOpen      extends SubscribeResultToClient
case object SubscribeResultToClientClosed                          extends SubscribeResultToClient
case class  SubscribeResultToClientMessages(messages: Seq[String]) extends SubscribeResultToClient
case object SubscribeResultToClientWaitForMessage                  extends SubscribeResultToClient

sealed trait NotificationToClient
case class  NotificationToClientMessage(message: String) extends NotificationToClient
case object NotificationToClientHeartbeat                extends NotificationToClient
case object NotificationToClientClosed                   extends NotificationToClient

object NonWebSocketSession {
  // The session must time out after 5 seconds of not having a receiving connection
  // http://sockjs.github.com/sockjs-protocol/sockjs-protocol-0.3.3.html#section-46
  private val TIMEOUT_CONNECTION = 5.seconds

  private val TIMEOUT_CONNECTION_MILLIS = TIMEOUT_CONNECTION.toMillis

  // The server must send a heartbeat frame every 25 seconds
  // http://sockjs.github.com/sockjs-protocol/sockjs-protocol-0.3.3.html#section-46
  private val TIMEOUT_HEARTBEAT = 25.seconds
}

/**
 * There should be at most one subscriber:
 * http://sockjs.github.com/sockjs-protocol/sockjs-protocol-0.3.3.html
 *
 * To avoid out of memory, the actor is stopped when there's no subscriber
 * for a long time. Timeout is also used to check if there's no message
 * for subscriber for a long time.
 * See TIMEOUT_CONNECTION and TIMEOUT_HEARTBEAT in NonWebSocketSessions.
 */
class NonWebSocketSession(var clientSubscriber: ActorRef, pathPrefix: String, action: Action) extends Actor {
  import NonWebSocketSession._

  private var sockJsHandler: SockJsHandler = _

  // Messages from handler to client are buffered here
  private val bufferForClientSubscriber = ArrayBuffer[String]()

  // ReceiveTimeout may not occurred if there's frequent Publish, thus we
  // need to manually check if there's no subscriber for a long time.
  // lastSubscribedAt must be Long to avoid Integer overflow, beacuse
  // System.currentTimeMillis() is used.
  private var lastSubscribedAt = 0L

  // Until the timeout occurs, the server must constantly serve
  // the close message
  private var closed = false

  override def preStart() {
    // sockJsHandler.onClose is called at postStop
    sockJsHandler = Routes.createSockJsHandler(pathPrefix)
    sockJsHandler.nonWebSocketSessionActorRef = self
    sockJsHandler.onOpen(action)

    lastSubscribedAt = System.currentTimeMillis()

    context.watch(clientSubscriber)  // Unsubscribed when stopped

    // Once set, the receive timeout stays in effect (i.e. continues firing
{'fraction_non_alphanumeric': 0.05672823218997362, 'fraction_numerical': 0.005863383172090296, 'mean_word_length': 4.107035928143713, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '4370086', 'n_tokens_mistral': 1830, 'n_tokens_neox': 1756, 'n_words': 574}