API challenge keepAliveStream flag

app/controllers/Challenge.scala

@@ -18,7 +18,8 @@ import lila.socket.Socket.SocketVersion
 import lila.user.{ User => UserModel }
 
 final class Challenge(
-    env: Env
+    env: Env,
+    apiC: Api
 ) extends LilaController(env) {
 
   def api = env.challenge.api
@@ -336,10 +337,13 @@ final class Challenge(
                           case _ =>
                             env.challenge.api create challenge map {
                               case true =>
-                                JsonOk(
-                                  env.challenge.jsonView
-                                    .show(challenge, SocketVersion(0), lila.challenge.Direction.Out.some)
-                                )
+                                val json = env.challenge.jsonView
+                                  .show(challenge, SocketVersion(0), lila.challenge.Direction.Out.some)
+                                if (config.keepAliveStream)
+                                  apiC.sourceToNdJsonOption(
+                                    apiC.addKeepAlive(env.challenge.keepAliveStream(challenge, json))
+                                  )
+                                else JsonOk(json)
                               case false =>
                                 BadRequest(jsonError("Challenge not created"))
                             }

modules/challenge/src/main/ChallengeKeepAliveStream.scala

@@ -0,0 +1,32 @@
+package lila.challenge
+
+import akka.stream.scaladsl._
+import play.api.libs.json._
+import scala.concurrent.duration._
+
+import lila.common.Bus
+
+final class ChallengeKeepAliveStream(api: ChallengeApi)(implicit
+    ec: scala.concurrent.ExecutionContext,
+    scheduler: akka.actor.Scheduler
+) {
+  def apply(challenge: Challenge, initialJson: JsObject): Source[JsValue, _] =
+    Source(List(initialJson)) concat
+      Source.queue[JsObject](1, akka.stream.OverflowStrategy.dropHead).mapMaterializedValue { queue =>
+        val keepAliveInterval = scheduler.scheduleWithFixedDelay(15 seconds, 15 seconds) { () =>
+          api.ping(challenge.id).unit
+        }
+        def completeWith(msg: String) = {
+          queue.offer(Json.obj("done" -> msg)) >>- queue.complete()
+        }.unit
+        val sub = Bus.subscribeFun("challenge") {
+          case Event.Accept(c, _) if c.id == challenge.id => completeWith("accepted")
+          case Event.Cancel(c) if c.id == challenge.id    => completeWith("canceled")
+          case Event.Decline(c) if c.id == challenge.id   => completeWith("declined")
+        }
+        queue.watchCompletion().foreach { _ =>
+          keepAliveInterval.cancel()
+          Bus.unsubscribe(sub, "challenge")
+        }
+      }
+}

modules/challenge/src/main/Env.scala

@@ -27,6 +27,7 @@ final class Env(
 )(implicit
     ec: scala.concurrent.ExecutionContext,
     system: akka.actor.ActorSystem,
+    scheduler: akka.actor.Scheduler,
     mode: play.api.Mode
 ) {
 
@@ -55,6 +56,8 @@ final class Env(
 
   lazy val msg = wire[ChallengeMsg]
 
+  lazy val keepAliveStream = wire[ChallengeKeepAliveStream]
+
   val forms = new ChallengeForm
 
   system.scheduler.scheduleWithFixedDelay(10 seconds, 3343 millis) { () =>

modules/setup/src/main/ApiConfig.scala

@@ -19,7 +19,8 @@ final case class ApiConfig(
     color: Color,
     position: Option[FEN] = None,
     acceptByToken: Option[String] = None,
-    message: Option[Template]
+    message: Option[Template],
+    keepAliveStream: Boolean
 ) {
 
   def perfType: Option[PerfType] = PerfPicker.perfType(chess.Speed(clock), variant, days)
@@ -52,7 +53,8 @@ object ApiConfig extends BaseHumanConfig {
       c: Option[String],
       pos: Option[FEN],
       tok: Option[String],
-      msg: Option[String]
+      msg: Option[String],
+      keepAliveStream: Option[Boolean]
   ) =
     new ApiConfig(
       variant = chess.variant.Variant.orDefault(~v),
@@ -62,7 +64,8 @@ object ApiConfig extends BaseHumanConfig {
       color = Color.orDefault(~c),
       position = pos,
       acceptByToken = tok,
-      message = msg map Template
+      message = msg map Template,
+      keepAliveStream = ~keepAliveStream
     ).autoVariant
 
   def validFen(variant: Variant, fen: Option[FEN]) =

modules/setup/src/main/SetupForm.scala

@@ -160,12 +160,13 @@ object SetupForm {
       mapping(
         variant,
         clock,
-        "days"          -> optional(days),
-        "rated"         -> boolean,
-        "color"         -> optional(color),
-        "fen"           -> fenField,
-        "acceptByToken" -> optional(nonEmptyText),
-        "message"       -> message
+        "days"            -> optional(days),
+        "rated"           -> boolean,
+        "color"           -> optional(color),
+        "fen"             -> fenField,
+        "acceptByToken"   -> optional(nonEmptyText),
+        "message"         -> message,
+        "keepAliveStream" -> optional(boolean)
       )(ApiConfig.from)(_ => none)
         .verifying("invalidFen", _.validFen)
         .verifying("can't be rated", _.validRated)