-
Notifications
You must be signed in to change notification settings - Fork 595
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
add java api for websocket testkit #21184
And additionally adds unit test for WebSocketDirectives #20466
- Loading branch information
1 parent
7b030bf
commit 75dc20b
Showing
12 changed files
with
332 additions
and
19 deletions.
There are no files selected for viewing
121 changes: 121 additions & 0 deletions
121
...cs/rst/java/code/docs/http/javadsl/server/directives/WebSocketDirectivesExamplesTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,121 @@ | ||
/* | ||
* Copyright (C) 2016-2016 Lightbend Inc. <http://www.lightbend.com> | ||
*/ | ||
package docs.http.javadsl.server.directives; | ||
|
||
import akka.NotUsed; | ||
import akka.http.javadsl.model.HttpRequest; | ||
import akka.http.javadsl.model.StatusCodes; | ||
import akka.http.javadsl.model.Uri; | ||
import akka.http.javadsl.model.headers.SecWebSocketProtocol; | ||
import akka.http.javadsl.model.ws.BinaryMessage; | ||
import akka.http.javadsl.model.ws.Message; | ||
import akka.http.javadsl.model.ws.TextMessage; | ||
import akka.http.javadsl.server.Route; | ||
import akka.http.javadsl.testkit.JUnitRouteTest; | ||
import akka.http.javadsl.testkit.WSProbe; | ||
import akka.stream.OverflowStrategy; | ||
import akka.stream.javadsl.Flow; | ||
import akka.stream.javadsl.Sink; | ||
import akka.stream.javadsl.Source; | ||
import akka.util.ByteString; | ||
import org.junit.Test; | ||
import scala.concurrent.duration.FiniteDuration; | ||
|
||
import java.util.Arrays; | ||
import java.util.Collections; | ||
import java.util.concurrent.TimeUnit; | ||
|
||
public class WebSocketDirectivesExamplesTest extends JUnitRouteTest { | ||
|
||
@Test | ||
public void testHandleWebSocketMessages() { | ||
//#handleWebSocketMessages | ||
final Flow<Message, Message, NotUsed> greeter = Flow.of(Message.class).mapConcat(msg -> { | ||
if (msg instanceof TextMessage) { | ||
final TextMessage tm = (TextMessage) msg; | ||
final TextMessage ret = TextMessage.create(Source.single("Hello ").concat(tm.getStreamedText()).concat(Source.single("!"))); | ||
return Collections.singletonList(ret); | ||
} else if (msg instanceof BinaryMessage) { | ||
final BinaryMessage bm = (BinaryMessage) msg; | ||
bm.getStreamedData().runWith(Sink.ignore(), materializer()); | ||
return Collections.emptyList(); | ||
} else { | ||
throw new IllegalArgumentException("Unsupported message type!"); | ||
} | ||
}); | ||
|
||
final Route websocketRoute = path("greeter", () -> | ||
handleWebSocketMessages(greeter) | ||
); | ||
|
||
// create a testing probe representing the client-side | ||
final WSProbe wsClient = WSProbe.create(system(), materializer()); | ||
|
||
// WS creates a WebSocket request for testing | ||
testRoute(websocketRoute).run(WS(Uri.create("/greeter"), wsClient.flow(), materializer())) | ||
.assertStatusCode(StatusCodes.SWITCHING_PROTOCOLS); | ||
|
||
// manually run a WS conversation | ||
wsClient.sendMessage("Peter"); | ||
wsClient.expectMessage("Hello Peter!"); | ||
|
||
wsClient.sendMessage(BinaryMessage.create(ByteString.fromString("abcdef"))); | ||
wsClient.expectNoMessage(FiniteDuration.create(100, TimeUnit.MILLISECONDS)); | ||
|
||
wsClient.sendMessage("John"); | ||
wsClient.expectMessage("Hello John!"); | ||
|
||
wsClient.sendCompletion(); | ||
wsClient.expectCompletion(); | ||
//#handleWebSocketMessages | ||
} | ||
|
||
@Test | ||
public void testHandleWebSocketMessagesForProtocol() { | ||
//#handleWebSocketMessagesForProtocol | ||
final Flow<Message, Message, NotUsed> greeterService = Flow.of(Message.class).mapConcat(msg -> { | ||
if (msg instanceof TextMessage) { | ||
final TextMessage tm = (TextMessage) msg; | ||
final TextMessage ret = TextMessage.create(Source.single("Hello ").concat(tm.getStreamedText()).concat(Source.single("!"))); | ||
return Collections.singletonList(ret); | ||
} else if (msg instanceof BinaryMessage) { | ||
final BinaryMessage bm = (BinaryMessage) msg; | ||
bm.getStreamedData().runWith(Sink.ignore(), materializer()); | ||
return Collections.emptyList(); | ||
} else { | ||
throw new IllegalArgumentException("Unsupported message type!"); | ||
} | ||
}); | ||
|
||
final Flow<Message, Message, NotUsed> echoService = Flow.of(Message.class).buffer(1, OverflowStrategy.backpressure()); | ||
|
||
final Route websocketMultipleProtocolRoute = path("services", () -> | ||
route( | ||
handleWebSocketMessagesForProtocol(greeterService, "greeter"), | ||
handleWebSocketMessagesForProtocol(echoService, "echo") | ||
) | ||
); | ||
|
||
// create a testing probe representing the client-side | ||
final WSProbe wsClient = WSProbe.create(system(), materializer()); | ||
|
||
// WS creates a WebSocket request for testing | ||
testRoute(websocketMultipleProtocolRoute) | ||
.run(WS(Uri.create("/services"), wsClient.flow(), materializer(), Arrays.asList("other", "echo"))) | ||
.assertHeaderExists(SecWebSocketProtocol.create("echo")); | ||
|
||
wsClient.sendMessage("Peter"); | ||
wsClient.expectMessage("Peter"); | ||
|
||
wsClient.sendMessage(BinaryMessage.create(ByteString.fromString("abcdef"))); | ||
wsClient.expectMessage(ByteString.fromString("abcdef")); | ||
|
||
wsClient.sendMessage("John"); | ||
wsClient.expectMessage("John"); | ||
|
||
wsClient.sendCompletion(); | ||
wsClient.expectCompletion(); | ||
//#handleWebSocketMessagesForProtocol | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
20 changes: 20 additions & 0 deletions
20
akka-http-core/src/main/java/akka/http/javadsl/model/headers/SecWebSocketProtocol.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,20 @@ | ||
/** | ||
* Copyright (C) 2016-2016 Lightbend Inc. <http://www.lightbend.com> | ||
*/ | ||
|
||
package akka.http.javadsl.model.headers; | ||
|
||
import akka.http.impl.util.Util; | ||
|
||
/** | ||
* Model for the `Sec-WebSocket-Protocol` header. | ||
*/ | ||
public abstract class SecWebSocketProtocol extends akka.http.scaladsl.model.HttpHeader { | ||
public abstract Iterable<String> getProtocols(); | ||
|
||
public static SecWebSocketProtocol create(String... protocols) { | ||
return new akka.http.scaladsl.model.headers.Sec$minusWebSocket$minusProtocol(Util.convertArray(protocols)); | ||
} | ||
|
||
} | ||
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
108 changes: 108 additions & 0 deletions
108
akka-http-testkit/src/main/scala/akka/http/javadsl/testkit/WSProbe.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,108 @@ | ||
/* | ||
* Copyright (C) 2016-2016 Lightbend Inc. <http://www.lightbend.com> | ||
*/ | ||
|
||
package akka.http.javadsl.testkit | ||
|
||
import akka.NotUsed | ||
import akka.actor.ActorSystem | ||
import akka.http.javadsl.model.ws.Message | ||
import akka.stream.Materializer | ||
import akka.stream.javadsl.Flow | ||
import akka.stream.scaladsl | ||
import akka.util.ByteString | ||
|
||
import akka.http.scaladsl.{ testkit ⇒ st } | ||
|
||
import akka.http.impl.util.JavaMapping.Implicits._ | ||
|
||
import scala.concurrent.duration._ | ||
|
||
/** | ||
* A WSProbe is a probe that implements a `Flow[Message, Message, Unit]` for testing | ||
* websocket code. | ||
* | ||
* Requesting elements is handled automatically. | ||
*/ | ||
class WSProbe(delegate: st.WSProbe) { | ||
|
||
def flow: Flow[Message, Message, Any] = { | ||
val underlying = scaladsl.Flow[Message].map(_.asScala).via(delegate.flow).map(_.asJava) | ||
new Flow[Message, Message, NotUsed](underlying) | ||
} | ||
|
||
/** | ||
* Send the given messages out of the flow. | ||
*/ | ||
def sendMessage(message: Message): Unit = delegate.sendMessage(message.asScala) | ||
|
||
/** | ||
* Send a text message containing the given string out of the flow. | ||
*/ | ||
def sendMessage(text: String): Unit = delegate.sendMessage(text) | ||
|
||
/** | ||
* Send a binary message containing the given bytes out of the flow. | ||
*/ | ||
def sendMessage(bytes: ByteString): Unit = delegate.sendMessage(bytes) | ||
|
||
/** | ||
* Complete the output side of the flow. | ||
*/ | ||
def sendCompletion(): Unit = delegate.sendCompletion() | ||
|
||
/** | ||
* Expect a message on the input side of the flow. | ||
*/ | ||
def expectMessage(): Message = delegate.expectMessage() | ||
|
||
/** | ||
* Expect a text message on the input side of the flow and compares its payload with the given one. | ||
* If the received message is streamed its contents are collected and then asserted against the given | ||
* String. | ||
*/ | ||
def expectMessage(text: String): Unit = delegate.expectMessage(text) | ||
|
||
/** | ||
* Expect a binary message on the input side of the flow and compares its payload with the given one. | ||
* If the received message is streamed its contents are collected and then asserted against the given | ||
* ByteString. | ||
*/ | ||
def expectMessage(bytes: ByteString): Unit = delegate.expectMessage(bytes) | ||
|
||
/** | ||
* Expect no message on the input side of the flow. | ||
*/ | ||
def expectNoMessage(): Unit = delegate.expectNoMessage() | ||
|
||
/** | ||
* Expect no message on the input side of the flow for the given maximum duration. | ||
*/ | ||
def expectNoMessage(max: FiniteDuration): Unit = delegate.expectNoMessage(max) | ||
|
||
/** | ||
* Expect completion on the input side of the flow. | ||
*/ | ||
def expectCompletion(): Unit = delegate.expectCompletion() | ||
|
||
} | ||
|
||
object WSProbe { | ||
|
||
// A convenient method to create WSProbe with default maxChunks and maxChunkCollectionMills | ||
def create(system: ActorSystem, materializer: Materializer): WSProbe = { | ||
create(system, materializer, 1000, 5000) | ||
} | ||
|
||
/** | ||
* Creates a WSProbe to use in tests against websocket handlers. | ||
* | ||
* @param maxChunks The maximum number of chunks to collect for streamed messages. | ||
* @param maxChunkCollectionMills The maximum time in milliseconds to collect chunks for streamed messages. | ||
*/ | ||
def create(system: ActorSystem, materializer: Materializer, maxChunks: Int, maxChunkCollectionMills: Long): WSProbe = { | ||
val delegate = st.WSProbe(maxChunks, maxChunkCollectionMills)(system, materializer) | ||
new WSProbe(delegate) | ||
} | ||
|
||
} |
Oops, something went wrong.