=doc add more documentation for WebsocketDirectives and testing

akka-docs-dev/rst/scala/code/docs/http/scaladsl/server/WebsocketExampleSpec.scala

@@ -4,6 +4,8 @@
 
 package docs.http.scaladsl.server
 
+import akka.http.scaladsl.model.ws.BinaryMessage
+import akka.stream.scaladsl.Sink
 import org.scalatest.{ Matchers, WordSpec }
 
 class WebsocketExampleSpec extends WordSpec with Matchers {
@@ -27,13 +29,16 @@ class WebsocketExampleSpec extends WordSpec with Matchers {
     // returns a greeting message for that name
     val greeterWebsocketService =
       Flow[Message]
-        .collect {
+        .mapConcat {
           // we match but don't actually consume the text message here,
           // rather we simply stream it back as the tail of the response
           // this means we might start sending the response even before the
           // end of the incoming message has been received
-          case tm: TextMessage ⇒ TextMessage(Source.single("Hello ") ++ tm.textStream)
-          // ignore binary messages
+          case tm: TextMessage ⇒ TextMessage(Source.single("Hello ") ++ tm.textStream) :: Nil
+          case bm: BinaryMessage =>
+            // ignore binary messages but drain content to avoid the stream being clogged
+            bm.dataStream.runWith(Sink.ignore)
+            Nil
         }
     //#websocket-handler
 

akka-docs-dev/rst/scala/code/docs/http/scaladsl/server/directives/WebsocketDirectivesExamplesSpec.scala

@@ -0,0 +1,104 @@
+/*
+ * Copyright (C) 2009-2015 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package docs.http.scaladsl.server.directives
+
+import scala.concurrent.duration._
+
+import akka.util.ByteString
+
+import akka.stream.OverflowStrategy
+import akka.stream.scaladsl.{ Sink, Source, Flow }
+
+import docs.http.scaladsl.server.RoutingSpec
+
+import akka.http.scaladsl.model.ws.{ TextMessage, Message, BinaryMessage }
+import akka.http.scaladsl.testkit.WSProbe
+
+class WebsocketDirectivesExamplesSpec extends RoutingSpec {
+  "greeter-service" in {
+    def greeter: Flow[Message, Message, Any] =
+      Flow[Message].mapConcat {
+        case tm: TextMessage ⇒
+          TextMessage(Source.single("Hello ") ++ tm.textStream ++ Source.single("!")) :: Nil
+        case bm: BinaryMessage ⇒
+          // ignore binary messages but drain content to avoid the stream being clogged
+          bm.dataStream.runWith(Sink.ignore)
+          Nil
+      }
+    val websocketRoute =
+      path("greeter") {
+        handleWebsocketMessages(greeter)
+      }
+
+    // create a testing probe representing the client-side
+    val wsClient = WSProbe()
+
+    // WS creates a Websocket request for testing
+    WS("/greeter", wsClient.flow) ~> websocketRoute ~>
+      check {
+        // check response for WS Upgrade headers
+        isWebsocketUpgrade shouldEqual true
+
+        // manually run a WS conversation
+        wsClient.sendMessage("Peter")
+        wsClient.expectMessage("Hello Peter!")
+
+        wsClient.sendMessage(BinaryMessage(ByteString("abcdef")))
+        wsClient.expectNoMessage(100.millis)
+
+        wsClient.sendMessage("John")
+        wsClient.expectMessage("Hello John!")
+
+        wsClient.sendCompletion()
+        wsClient.expectCompletion()
+      }
+  }
+
+  "handle-multiple-protocols" in {
+    def greeterService: Flow[Message, Message, Any] =
+      Flow[Message].mapConcat {
+        case tm: TextMessage ⇒
+          TextMessage(Source.single("Hello ") ++ tm.textStream ++ Source.single("!")) :: Nil
+        case bm: BinaryMessage ⇒
+          // ignore binary messages but drain content to avoid the stream being clogged
+          bm.dataStream.runWith(Sink.ignore)
+          Nil
+      }
+
+    def echoService: Flow[Message, Message, Any] =
+      Flow[Message]
+        // needed because a noop flow hasn't any buffer that would start processing in tests
+        .buffer(1, OverflowStrategy.backpressure)
+
+    def websocketMultipleProtocolRoute =
+      path("services") {
+        handleWebsocketMessagesForProtocol(greeterService, "greeter") ~
+          handleWebsocketMessagesForProtocol(echoService, "echo")
+      }
+
+    val wsClient = WSProbe()
+
+    // WS creates a Websocket request for testing
+    WS("/services", wsClient.flow, List("other", "echo")) ~>
+      websocketMultipleProtocolRoute ~>
+      check {
+        expectWebsocketUpgradeWithProtocol { protocol ⇒
+          protocol shouldEqual "echo"
+
+          wsClient.sendMessage("Peter")
+          wsClient.expectMessage("Peter")
+
+          wsClient.sendMessage(BinaryMessage(ByteString("abcdef")))
+          wsClient.expectMessage(ByteString("abcdef"))
+
+          wsClient.sendMessage("John")
+          wsClient.expectMessage("John")
+
+          wsClient.sendCompletion()
+          wsClient.expectCompletion()
+        }
+      }
+  }
+}