improve socket communications
parent
785bca0d01
commit
c303db1ac4
|
@ -16,7 +16,7 @@ object Main extends LilaController {
|
||||||
|
|
||||||
def websocket = Socket { implicit ctx =>
|
def websocket = Socket { implicit ctx =>
|
||||||
get("sri") ?? { uid =>
|
get("sri") ?? { uid =>
|
||||||
Env.site.socketHandler(uid, ctx.userId, get("flag"), getBool("tv"))
|
Env.site.socketHandler(uid, ctx.userId, get("flag"))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -24,7 +24,7 @@ object Round extends LilaController with TheftPrevention {
|
||||||
|
|
||||||
def websocketWatcher(gameId: String, color: String) = Socket[JsValue] { implicit ctx =>
|
def websocketWatcher(gameId: String, color: String) = Socket[JsValue] { implicit ctx =>
|
||||||
(get("sri") |@| getInt("version")).tupled ?? {
|
(get("sri") |@| getInt("version")).tupled ?? {
|
||||||
case (uid, version) => env.socketHandler.watcher(gameId, color, version, uid, ctx.me, ctx.ip, getBool("tv"))
|
case (uid, version) => env.socketHandler.watcher(gameId, color, version, uid, ctx.me, ctx.ip)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -18,7 +18,7 @@ final class Bus(system: ActorSystem) extends Extension with EventBus {
|
||||||
* @return true if successful and false if not (because it was already subscribed to that Classifier, or otherwise)
|
* @return true if successful and false if not (because it was already subscribed to that Classifier, or otherwise)
|
||||||
*/
|
*/
|
||||||
def subscribe(subscriber: Subscriber, to: Classifier): Boolean = {
|
def subscribe(subscriber: Subscriber, to: Classifier): Boolean = {
|
||||||
log(s"subscribe $to $subscriber")
|
// log(s"subscribe $to $subscriber")
|
||||||
bus.subscribe(subscriber, to)
|
bus.subscribe(subscriber, to)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -32,7 +32,7 @@ final class Bus(system: ActorSystem) extends Extension with EventBus {
|
||||||
* @return true if successful and false if not (because it wasn't subscribed to that Classifier, or otherwise)
|
* @return true if successful and false if not (because it wasn't subscribed to that Classifier, or otherwise)
|
||||||
*/
|
*/
|
||||||
def unsubscribe(subscriber: Subscriber, from: Classifier): Boolean = {
|
def unsubscribe(subscriber: Subscriber, from: Classifier): Boolean = {
|
||||||
log(s"[UN]subscribe $from $subscriber")
|
// log(s"[UN]subscribe $from $subscriber")
|
||||||
bus.unsubscribe(subscriber, from)
|
bus.unsubscribe(subscriber, from)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -40,7 +40,7 @@ final class Bus(system: ActorSystem) extends Extension with EventBus {
|
||||||
* Attempts to deregister the subscriber from all Classifiers it may be subscribed to
|
* Attempts to deregister the subscriber from all Classifiers it may be subscribed to
|
||||||
*/
|
*/
|
||||||
def unsubscribe(subscriber: Subscriber) {
|
def unsubscribe(subscriber: Subscriber) {
|
||||||
log(s"[UN]subscribe ALL $subscriber")
|
// log(s"[UN]subscribe ALL $subscriber")
|
||||||
bus unsubscribe subscriber
|
bus unsubscribe subscriber
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -48,11 +48,11 @@ final class Bus(system: ActorSystem) extends Extension with EventBus {
|
||||||
* Publishes the specified Event to this bus
|
* Publishes the specified Event to this bus
|
||||||
*/
|
*/
|
||||||
def publish(event: Event) {
|
def publish(event: Event) {
|
||||||
log(event.toString)
|
// log(event.toString)
|
||||||
bus publish event
|
bus publish event
|
||||||
}
|
}
|
||||||
|
|
||||||
private def log(msg: String) {
|
private def log(msg: => String) {
|
||||||
// loginfo(msg)
|
// loginfo(msg)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -37,7 +37,7 @@ final class Featured(
|
||||||
oneId = game.id.some
|
oneId = game.id.some
|
||||||
rendererActor ? actorApi.RenderFeaturedJs(game) onSuccess {
|
rendererActor ? actorApi.RenderFeaturedJs(game) onSuccess {
|
||||||
case html: Html =>
|
case html: Html =>
|
||||||
bus.publish(lila.hub.actorApi.game.ChangeFeatured(game.id, html), 'changeFeaturedGame)
|
bus.publish(lila.hub.actorApi.game.ChangeFeatured(html), 'changeFeaturedGame)
|
||||||
}
|
}
|
||||||
GameRepo setTv game.id
|
GameRepo setTv game.id
|
||||||
}
|
}
|
||||||
|
|
|
@ -95,7 +95,7 @@ case class Propagate(data: Atom, propagations: List[Propagation] = Nil) {
|
||||||
}
|
}
|
||||||
|
|
||||||
package game {
|
package game {
|
||||||
case class ChangeFeatured(id: String, html: Html)
|
case class ChangeFeatured(html: Html)
|
||||||
case object Count
|
case object Count
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -24,6 +24,8 @@ private[lobby] final class Socket(
|
||||||
router: akka.actor.ActorSelection,
|
router: akka.actor.ActorSelection,
|
||||||
uidTtl: Duration) extends SocketActor[Member](uidTtl) with Historical[Member] {
|
uidTtl: Duration) extends SocketActor[Member](uidTtl) with Historical[Member] {
|
||||||
|
|
||||||
|
context.system.lilaBus.subscribe(self, 'changeFeaturedGame)
|
||||||
|
|
||||||
def receiveSpecific = {
|
def receiveSpecific = {
|
||||||
|
|
||||||
case PingVersion(uid, v) => {
|
case PingVersion(uid, v) => {
|
||||||
|
|
|
@ -36,7 +36,6 @@ private[lobby] final class SocketHandler(
|
||||||
val join = Join(uid = uid, user = user)
|
val join = Join(uid = uid, user = user)
|
||||||
Handler(hub, socket, uid, join, user map (_.id)) {
|
Handler(hub, socket, uid, join, user map (_.id)) {
|
||||||
case Connected(enum, member) =>
|
case Connected(enum, member) =>
|
||||||
member.setTv
|
|
||||||
controller(socket, uid, member) -> enum
|
controller(socket, uid, member) -> enum
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -57,22 +57,27 @@ final class Env(
|
||||||
def receive = actorMapReceive
|
def receive = actorMapReceive
|
||||||
}), name = ActorMapName)
|
}), name = ActorMapName)
|
||||||
|
|
||||||
private val socketHub = system.actorOf(
|
private val socketHub = {
|
||||||
Props(new lila.socket.SocketHubActor[Socket] {
|
val actor = system.actorOf(
|
||||||
def mkActor(id: String) = new Socket(
|
Props(new lila.socket.SocketHubActor[Socket] {
|
||||||
gameId = id,
|
def mkActor(id: String) = new Socket(
|
||||||
history = history(),
|
gameId = id,
|
||||||
getUsername = getUsername,
|
history = history(),
|
||||||
uidTimeout = UidTimeout,
|
getUsername = getUsername,
|
||||||
socketTimeout = SocketTimeout,
|
uidTimeout = UidTimeout,
|
||||||
disconnectTimeout = PlayerDisconnectTimeout,
|
socketTimeout = SocketTimeout,
|
||||||
ragequitTimeout = PlayerRagequitTimeout)
|
disconnectTimeout = PlayerDisconnectTimeout,
|
||||||
def receive: Receive = ({
|
ragequitTimeout = PlayerRagequitTimeout)
|
||||||
case msg@lila.chat.actorApi.ChatLine(id, line) =>
|
def receive: Receive = ({
|
||||||
self ! lila.hub.actorApi.map.Tell(id take 8, msg)
|
case msg@lila.chat.actorApi.ChatLine(id, line) =>
|
||||||
}: Receive) orElse socketHubReceive
|
self ! lila.hub.actorApi.map.Tell(id take 8, msg)
|
||||||
}),
|
case m: lila.hub.actorApi.game.ChangeFeatured => tellAll(m)
|
||||||
name = SocketName)
|
}: Receive) orElse socketHubReceive
|
||||||
|
}),
|
||||||
|
name = SocketName)
|
||||||
|
system.lilaBus.subscribe(actor, 'changeFeaturedGame)
|
||||||
|
actor
|
||||||
|
}
|
||||||
|
|
||||||
lazy val socketHandler = new SocketHandler(
|
lazy val socketHandler = new SocketHandler(
|
||||||
hub = hub,
|
hub = hub,
|
||||||
|
|
|
@ -85,10 +85,9 @@ private[round] final class SocketHandler(
|
||||||
version: Int,
|
version: Int,
|
||||||
uid: String,
|
uid: String,
|
||||||
user: Option[User],
|
user: Option[User],
|
||||||
ip: String,
|
ip: String): Fu[JsSocketHandler] =
|
||||||
tv: Boolean): Fu[JsSocketHandler] =
|
|
||||||
GameRepo.pov(gameId, colorName) flatMap {
|
GameRepo.pov(gameId, colorName) flatMap {
|
||||||
_ ?? { join(_, none, version, uid, "", user, ip, tv) }
|
_ ?? { join(_, none, version, uid, "", user, ip) }
|
||||||
}
|
}
|
||||||
|
|
||||||
def player(
|
def player(
|
||||||
|
@ -99,7 +98,7 @@ private[round] final class SocketHandler(
|
||||||
user: Option[User],
|
user: Option[User],
|
||||||
ip: String): Fu[JsSocketHandler] =
|
ip: String): Fu[JsSocketHandler] =
|
||||||
GameRepo.pov(fullId) flatMap {
|
GameRepo.pov(fullId) flatMap {
|
||||||
_ ?? { join(_, Some(Game takePlayerId fullId), version, uid, token, user, ip, false) }
|
_ ?? { join(_, Some(Game takePlayerId fullId), version, uid, token, user, ip) }
|
||||||
}
|
}
|
||||||
|
|
||||||
private def join(
|
private def join(
|
||||||
|
@ -109,8 +108,7 @@ private[round] final class SocketHandler(
|
||||||
uid: String,
|
uid: String,
|
||||||
token: String,
|
token: String,
|
||||||
user: Option[User],
|
user: Option[User],
|
||||||
ip: String,
|
ip: String): Fu[JsSocketHandler] = {
|
||||||
tv: Boolean): Fu[JsSocketHandler] = {
|
|
||||||
val join = Join(
|
val join = Join(
|
||||||
uid = uid,
|
uid = uid,
|
||||||
user = user,
|
user = user,
|
||||||
|
@ -121,7 +119,6 @@ private[round] final class SocketHandler(
|
||||||
socketHub ? Get(pov.gameId) mapTo manifest[ActorRef] flatMap { socket =>
|
socketHub ? Get(pov.gameId) mapTo manifest[ActorRef] flatMap { socket =>
|
||||||
Handler(hub, socket, uid, join, user map (_.id)) {
|
Handler(hub, socket, uid, join, user map (_.id)) {
|
||||||
case Connected(enum, member) =>
|
case Connected(enum, member) =>
|
||||||
if (tv) member.setTv
|
|
||||||
controller(pov.gameId, socket, uid, pov.ref, member) -> enum
|
controller(pov.gameId, socket, uid, pov.ref, member) -> enum
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -21,7 +21,7 @@ private final class TvBroadcast extends Actor {
|
||||||
|
|
||||||
case TvBroadcast.GetEnumerator => sender ! enumerator
|
case TvBroadcast.GetEnumerator => sender ! enumerator
|
||||||
|
|
||||||
case ChangeFeatured(_, html) =>
|
case ChangeFeatured(html) =>
|
||||||
channel push makeMessage("featured", Json.obj("html" -> html.toString))
|
channel push makeMessage("featured", Json.obj("html" -> html.toString))
|
||||||
|
|
||||||
case move: MoveEvent => channel push makeMessage("fen", Json.obj(
|
case move: MoveEvent => channel push makeMessage("fen", Json.obj(
|
||||||
|
|
|
@ -30,7 +30,7 @@ private[simulation] final class WatcherBot(
|
||||||
}
|
}
|
||||||
|
|
||||||
case Event(Some(game: lila.game.Game), _) => {
|
case Event(Some(game: lila.game.Game), _) => {
|
||||||
roundEnv.socketHandler.watcher(game.id, "white", 0, uid, user, ip, true) pipeTo self
|
roundEnv.socketHandler.watcher(game.id, "white", 0, uid, user, ip) pipeTo self
|
||||||
goto(TvConnect) using Id(game.id)
|
goto(TvConnect) using Id(game.id)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -61,7 +61,7 @@ private[simulation] final class WatcherBot(
|
||||||
|
|
||||||
case Event(Message("featured_id", obj), watcher: Watcher) => obj str "d" map { id =>
|
case Event(Message("featured_id", obj), watcher: Watcher) => obj str "d" map { id =>
|
||||||
watcher.channel.eofAndEnd()
|
watcher.channel.eofAndEnd()
|
||||||
roundEnv.socketHandler.watcher(id, "white", 0, uid, user, ip, true) pipeTo self
|
roundEnv.socketHandler.watcher(id, "white", 0, uid, user, ip) pipeTo self
|
||||||
goto(TvConnect) using Id(id)
|
goto(TvConnect) using Id(id)
|
||||||
} getOrElse stay
|
} getOrElse stay
|
||||||
|
|
||||||
|
|
|
@ -17,8 +17,7 @@ private[site] final class SocketHandler(
|
||||||
def apply(
|
def apply(
|
||||||
uid: String,
|
uid: String,
|
||||||
userId: Option[String],
|
userId: Option[String],
|
||||||
flag: Option[String],
|
flag: Option[String]): Fu[JsSocketHandler] = {
|
||||||
tv: Boolean): Fu[JsSocketHandler] = {
|
|
||||||
|
|
||||||
def controller: Handler.Controller = {
|
def controller: Handler.Controller = {
|
||||||
case ("liveGames", o) => o str "d" foreach { ids =>
|
case ("liveGames", o) => o str "d" foreach { ids =>
|
||||||
|
@ -28,7 +27,6 @@ private[site] final class SocketHandler(
|
||||||
|
|
||||||
Handler(hub, socket, uid, Join(uid, userId, flag), userId) {
|
Handler(hub, socket, uid, Join(uid, userId, flag), userId) {
|
||||||
case Connected(enum, member) =>
|
case Connected(enum, member) =>
|
||||||
if (tv) member.setTv
|
|
||||||
controller -> enum
|
controller -> enum
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -34,32 +34,34 @@ abstract class SocketActor[M <: SocketMember](uidTtl: Duration) extends Socket w
|
||||||
// generic message handler
|
// generic message handler
|
||||||
def receiveGeneric: Receive = {
|
def receiveGeneric: Receive = {
|
||||||
|
|
||||||
case Ping(uid) => ping(uid)
|
case Ping(uid) => ping(uid)
|
||||||
|
|
||||||
case Broom => broom
|
case Broom => broom
|
||||||
|
|
||||||
// when a member quits
|
// when a member quits
|
||||||
case Quit(uid) => quit(uid)
|
case Quit(uid) => quit(uid)
|
||||||
|
|
||||||
case NbMembers(nb) => pong = makePong(nb)
|
case NbMembers(nb) => pong = makePong(nb)
|
||||||
|
|
||||||
case WithUserIds(f) => f(userIds)
|
case WithUserIds(f) => f(userIds)
|
||||||
|
|
||||||
case GetUids => sender ! uids
|
case GetUids => sender ! uids
|
||||||
|
|
||||||
case LiveGames(uid, gameIds) => registerLiveGames(uid, gameIds)
|
case LiveGames(uid, gameIds) => registerLiveGames(uid, gameIds)
|
||||||
|
|
||||||
case move: MoveEvent => notifyMove(move)
|
case move: MoveEvent => notifyMove(move)
|
||||||
|
|
||||||
case SendTo(userId, msg) => sendTo(userId, msg)
|
case SendTo(userId, msg) => sendTo(userId, msg)
|
||||||
|
|
||||||
case SendTos(userIds, msg) => sendTos(userIds, msg)
|
case SendTos(userIds, msg) => sendTos(userIds, msg)
|
||||||
|
|
||||||
case Resync(uid) => resync(uid)
|
case Resync(uid) => resync(uid)
|
||||||
|
|
||||||
case Deploy(event, html) => notifyAll(makeMessage(event.key, html))
|
case Deploy(event, html) => notifyAll(makeMessage(event.key, html))
|
||||||
|
|
||||||
case ChangeFeatured(id, html) => notifyFeatured(id, html)
|
// the actor instance must subscribe to 'changeFeaturedGame to receive this message
|
||||||
|
// context.system.lilaBus.subscribe(self, 'changeFeaturedGame)
|
||||||
|
case ChangeFeatured(html) => notifyFeatured(html)
|
||||||
}
|
}
|
||||||
|
|
||||||
def receive = receiveSpecific orElse receiveGeneric
|
def receive = receiveSpecific orElse receiveGeneric
|
||||||
|
@ -173,10 +175,8 @@ abstract class SocketActor[M <: SocketMember](uidTtl: Duration) extends Socket w
|
||||||
members get uid foreach f
|
members get uid foreach f
|
||||||
}
|
}
|
||||||
|
|
||||||
private def notifyFeatured(id: String, html: Html) {
|
private def notifyFeatured(html: Html) {
|
||||||
val msg = makeMessage("featured", Json.obj("id" -> id, "html" -> html.toString))
|
val msg = makeMessage("featured", Json.obj("html" -> html.toString))
|
||||||
members.values foreach { m =>
|
members.values foreach { _.channel push msg }
|
||||||
if (m.hasTv) m.channel push msg
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -13,9 +13,7 @@ final class SocketHub extends Actor {
|
||||||
|
|
||||||
private val sockets = collection.mutable.Set[ActorRef]()
|
private val sockets = collection.mutable.Set[ActorRef]()
|
||||||
|
|
||||||
context.system.lilaBus.subscribe(self,
|
context.system.lilaBus.subscribe(self, 'moveEvent, 'users, 'deploy, 'nbMembers, 'socket)
|
||||||
'moveEvent, 'users, 'deploy, 'nbMembers, 'socket,
|
|
||||||
'changeFeaturedGame)
|
|
||||||
|
|
||||||
override def postStop() {
|
override def postStop() {
|
||||||
context.system.lilaBus.unsubscribe(self)
|
context.system.lilaBus.unsubscribe(self)
|
||||||
|
|
|
@ -8,15 +8,11 @@ trait SocketMember extends Ordered[SocketMember] {
|
||||||
|
|
||||||
// FIXME
|
// FIXME
|
||||||
private val privateLiveGames = collection.mutable.Set[String]()
|
private val privateLiveGames = collection.mutable.Set[String]()
|
||||||
private var privateHasTv: Boolean = false
|
|
||||||
|
|
||||||
def liveGames: Set[String] = privateLiveGames.toSet
|
def liveGames: Set[String] = privateLiveGames.toSet
|
||||||
|
|
||||||
def addLiveGames(ids: List[String]) { ids foreach privateLiveGames.+= }
|
def addLiveGames(ids: List[String]) { ids foreach privateLiveGames.+= }
|
||||||
|
|
||||||
def hasTv = privateHasTv
|
|
||||||
def setTv { privateHasTv = true }
|
|
||||||
|
|
||||||
def isAuth = userId.isDefined
|
def isAuth = userId.isDefined
|
||||||
|
|
||||||
def compare(other: SocketMember) = ~userId compare ~other.userId
|
def compare(other: SocketMember) = ~userId compare ~other.userId
|
||||||
|
|
|
@ -579,15 +579,7 @@ var storage = {
|
||||||
|
|
||||||
setTimeout(function() {
|
setTimeout(function() {
|
||||||
if (lichess.socket === null) {
|
if (lichess.socket === null) {
|
||||||
var socketSettings = lichess.socketDefaults;
|
lichess.socket = new strongSocket("/socket", 0, lichess.socketDefaults);
|
||||||
if ($('body').hasClass('embed-tv')) {
|
|
||||||
socketSettings.params.tv = 1;
|
|
||||||
socketSettings.events.featured = function changeFeatured(o) {
|
|
||||||
$('#featured_game').html(o.html);
|
|
||||||
$('body').trigger('lichess.content_loaded');
|
|
||||||
};
|
|
||||||
}
|
|
||||||
lichess.socket = new strongSocket("/socket", 0, socketSettings);
|
|
||||||
}
|
}
|
||||||
$(document).idleTimer(lichess.idleTime)
|
$(document).idleTimer(lichess.idleTime)
|
||||||
.on('idle.idleTimer', function() {
|
.on('idle.idleTimer', function() {
|
||||||
|
@ -928,8 +920,7 @@ var storage = {
|
||||||
name: "game"
|
name: "game"
|
||||||
},
|
},
|
||||||
params: {
|
params: {
|
||||||
ran: "--ranph--",
|
ran: "--ranph--"
|
||||||
tv: self.options.tv ? 1 : null
|
|
||||||
},
|
},
|
||||||
events: {
|
events: {
|
||||||
possible_moves: function(event) {
|
possible_moves: function(event) {
|
||||||
|
|
Loading…
Reference in New Issue