【三異步HTTP編程】 4. WebSockets

時間 2019-11-17

原文原文鏈接

WebSockets 是瀏覽器上的全雙工通訊協議。在WebSockets通道存在期間，客戶端和服務器之間能夠自由通訊。html

現代 HTML5 兼容的瀏覽器能夠經過 JavaScript API 原生地支持WebSockets。除了瀏覽器以外，還有許多WebSockets客戶端庫可用於服務器之間、原生的移動APP通訊等場景。在這些環境使用WebSockets的好處是能夠重用Play服務器現有的TCP端口。java

提示：到這裏查看支持WebSockets的瀏覽器相關問題。web

處理WebSockets

到目前爲止，咱們都是用 Action 來處理標準 HTTP 請求並返回標準 HTTP 響應。可是標準的 Action 並不能處理 WebSockets 這種徹底不一樣的請求。json

Play 的 WebSockets 功能創建在Akka stream的基礎上，將收到的 WebSockets 消息變成流，而後從流中產生響應併發送到客戶端。api

從概念上來講，一個「流」指收到消息、處理消息、最後產生消息這樣一種消息轉換。這裏的輸入和輸出能夠徹底解耦開來。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是最直觀的方式。websocket

這個例子中咱們發送的actor相似這樣：session

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
  }
}