【三 异步HTTP编程】 4. WebSockets

WebSockets 是浏览器上的全双工通讯协议。在WebSockets通道存在期间，客户端和服务器之间能够自由通讯。

现代 HTML5 兼容的浏览器能够经过 JavaScript API 原生地支持WebSockets。除了浏览器以外，还有许多WebSockets客户端库可用于服务器之间、原生的移动APP通讯等场景。在这些环境使用WebSockets的好处是能够重用Play服务器现有的TCP端口。

提示：到这里查看支持WebSockets的浏览器相关问题。

处理WebSockets

到目前为止，咱们都是用 Action 来处理标准 HTTP 请求并返回标准 HTTP 响应。可是标准的 Action 并不能处理 WebSockets 这种彻底不一样的请求。

Play 的 WebSockets 功能创建在Akka stream的基础上，将收到的 WebSockets 消息变成流，而后从流中产生响应并发送到客户端。

从概念上来讲，一个 “流” 指收到消息、处理消息、最后产生消息这样一种消息转换。这里的输入和输出能够彻底解耦开来。Akka提供了 Flow.fromSinkAndSource 构造函数来处理这种场景，事实上处理WebSockets时，输入和输出并不直接相互链接。

Play在 WebSocket 类中提供了构造WebSockets的工厂方法。

使用 Akka Streams 及 actors

为了使用 actor 来处理WebSockets，咱们使用Play提供的ActorFlow工具来将ActorRef转换为流。当Play接收到一个WebSockets链接时，会建立一个actor，它接受一个 ActorRef => akka.actor.Props 函数为参数并返回一个socket：

import play.api.mvc._
import play.api.libs.streams.ActorFlow
import javax.inject.Inject
import akka.actor.ActorSystem
import akka.stream.Materializer

class Application @Inject()(cc:ControllerComponents)(implicit system: ActorSystem, mat: Materializer) extends AbstractController(cc) {

  def socket = WebSocket.accept[String, String] { request =>
    ActorFlow.actorRef { out =>
      MyWebSocketActor.props(out)
    }
  }
}

注意ActorFlow.actorRef(...) 能够用 Flow[In, Out, _] 替换，可是使用actor是最直观的方式。

这个例子中咱们发送的actor相似这样：

import akka.actor._

object MyWebSocketActor {
    def props(out: ActorRef) = Props(new MyWebSocketActor(out))
}

class MyWebSocketActor(out: ActorRef) extends Actor {
    def receive = {
        case msg: String =>
            out ! ("I received your message: " + msg)
    }
}

从客户端接收到的全部消息都会被发往actor，而 Play 提供给actor的全部消息都会被发往客户端。上边的代码中，actor仅仅将收到的消息加上 “I received your message: ” 前缀而后发回去。

检测WebSocket什么时候关闭

当WebSocket关闭时，Play将自动中止actor。就是说你能够经过实现actor的postStop方法来作一些清理工做，如清理WebSocket用到的资源。如：

override def postStop() = {
    someResource.close()
}

关闭WebSocket

在actor中止时，Play也将自动关闭其处理的WebSocket。所以要手动关闭WebSocket，能够主动向actor发送PoisonPill：

impoort akka.actor.PoisonPill

self ! PoisonPill

拒绝WebSocket

某些时候咱们须要拒绝一个WebSocket请求，如：链接前须要先对用户鉴权，或者请求了不存在的资源。Play提供了 acceptOrResult方法来应对这种状况，你能够直接返回一个Result（如 FORBIDDEN、NOT FOUND 等），也能够返回一个处理WebSocket的actor：

import play.api.mvc._
import play.api.libs.streams.ActorFlow
import javax.inject.Inject
import akka.actor.ActorSystem
import akka.stream.Materializer

class Application @Inject()(cc: ControllerComponents)(implicit system: ActorSystem, mat: Materializer) extends AbstractController(cc) {
    
    def socket = WebSocket.acceptOrResult[String, String] { request =>
        Future.successful(request.session.get("user") match {
            case None => Left(Forbidden)
            case Some(_) => Right(ActorFlow.actorRef {
                MyWebSOcketActor.props(out)
            })
        })
    }
}

注意：WebSocket协议并未实现同源策略，所以没法防护跨站点WebSocket劫持。要保护websocket不被劫持，须要根据server的origin来检测request的Origin头，而后手动来进行鉴权（包括CSRF token）。若是一个WebSocket没有经过安全性检查，能够直接用acceptOrResult方法返回FORBIDDEN。

处理不一样类型的消息

如今咱们只处理了String类型的数据。其实Play也内置了 Array[Byte] 的handler，并且能够从String类型的数据帧中解析出JsValue。数据类型能够在WebSocket的建立方法中以类型参数形式来定义：

import play.api.libs.json._
import play.api.mvc._
import play.api.libs.streams.ActorFlow
import javax.inject.Inject
import akka.actor.ActorSystem
import akka.stream.Materializer

class Application @Inject()(cc:ControllerComponents)
                           (implicit system: ActorSystem, mat: Materializer)
  extends AbstractController(cc) {

  def socket = WebSocket.accept[JsValue, JsValue] { request =>
    ActorFlow.actorRef { out =>
      MyWebSocketActor.props(out)
    }
  }
}

你可能注意到了上边的两个JsValue类型，它容许咱们处理不一样类型的输入及输出。在高层级的数据帧类型上尤为有用。

举个栗子，好比咱们但愿收到JSON数据类型，并将输入的消息转为InEvent对象，而后将输出消息格式化为OutEvent对象。首先须要建立JSON来格式化咱们的InEvent及OutEvent：

import play.api.libs.json._

implicit val inEventFormat = Json.format[InEvent]
implicit val outEventFormat = Json.format[OutEvent]

而后能够为这些类型来建立WebSocket MessageFlowTransformer：

import play.api.mvc.WebSocket.MessageFlowTransformer

implicit val messageFlowTransformer = MessageFlowTransformer.jsonMessageFlowTransformer[InEvent, OutEvent]

最后在WebSocket中使用它们：

import play.api.mvc._

import play.api.libs.streams.ActorFlow
import javax.inject.Inject
import akka.actor.ActorSystem
import akka.stream.Materializer

class Application @Inject()(cc:ControllerComponents)
                           (implicit system: ActorSystem, mat: Materializer)
  extends AbstractController(cc) {

  def socket = WebSocket.accept[InEvent, OutEvent] { request =>
    ActorFlow.actorRef { out =>
      MyWebSocketActor.props(out)
    }
  }
}

如今咱们的actor能够直接受到InEvent类型的消息，而后直接发送 OutEvent。

使用Akka streams直接处理WebSockets

Actors抽象并非老是适合你的场景，特别是若是WebSockets自己表现得更像流的时候。

import play.api.mvc._
import akka.stream.scaladsl._

def socket = WebSocket.accept[String, String] { request =>

  // Log events to the console
  val in = Sink.foreach[String](println)

  // Send a single 'Hello!' message and then leave the socket open
  val out = Source.single("Hello!").concat(Source.maybe)

  Flow.fromSinkAndSource(in, out)
}

一个WebSocket能够访问初始化WebSocket链接的原始HTTP头，容许你检索标准头以及session数据。可是它不能访问请求体及HTTP响应。

在这个例子中，咱们建立了一个简单的 sink 来打印消息到控制台。并建立了一个简单的 source 来发送简单的 “Hello！”。咱们还须要维持一个永远不会发送任何内容的 source，不然咱们的单个source将终止流，从而终止掉链接。

提示：你能够在 https://www.websocket.org/echo.html 上测试WebSockets。只须要将 location 设置为： ws://localhsot:9000。

下面是一个丢弃输入数据，并简单返回 “Hello！”的例子：

import play.api.mvc._
import akka.stream.scaladsl._

def socket = WebSocket.accept[String, String] { request =>

  // Just ignore the input
  val in = Sink.ignore

  // Send a single 'Hello!' message and close
  val out = Source.single("Hello!")

  Flow.fromSinkAndSource(in, out)
}

下面是另外一个例子，将输入简单记录到标准输出，而后使用发送回client：

import play.api.mvc._
import akka.stream.scaladsl._

def socket =  WebSocket.accept[String, String] { request =>

  // log the message to stdout and send response back to client
  Flow[String].map { msg =>
    println(msg)
    "I received your message: " + msg
  }
}

设置WebSocket帧长度

你可使用play.server.websocket.frame.maxLength或者设置 --Dwebsocket.frame.maxLength系统变量来设置WebSocket数据帧的长度。举例以下：

sbt -Dwebsocket.frame.maxLength=64k run

你能够根据项目须要自由的调整适合的帧长度。同事使用较长的数据帧也能够减小DOS攻击。