streamalism.org

1. Overzicht

Als we willen dat onze webclients een dialoog onderhouden met onze server, dan kan WebSockets een handige oplossing zijn. WebSockets behouden een permanente full-duplex-verbinding. Dit geeft ons de mogelijkheid om bidirectionele berichten tussen onze server en client te verzenden.

In deze tutorial gaan we leren hoe we WebSockets kunnen gebruiken met Akka in het Play Framework.

2. Installatie

Laten we een eenvoudige chattoepassing opzetten. De gebruiker zal berichten naar de server sturen en de server zal reageren met een bericht van JSONPlaceholder.

2.1. De Play Framework-applicatie instellen

We bouwen deze applicatie met behulp van het Play Framework.

Laten we de instructies volgen van Inleiding tot spelen in Java om een ​​eenvoudige Play Framework-applicatie in te stellen en uit te voeren.

2.2. De benodigde JavaScript-bestanden toevoegen

We zullen ook met JavaScript moeten werken voor scripting aan de clientzijde. Hierdoor kunnen we nieuwe berichten ontvangen die door de server worden gepusht. Hiervoor gebruiken we de jQuery-bibliotheek.

Laten we jQuery toevoegen aan de onderkant van het app / views / index.scala.html het dossier:

2.3. Akka instellen

Ten slotte gebruiken we Akka om de WebSocket-verbindingen aan de serverzijde af te handelen.

Laten we naar het build.sbt bestand en voeg de afhankelijkheden toe.

We moeten de akka-acteur en akka-testkit afhankelijkheden:

libraryDependencies + = "com.typesafe.akka" %% "akka-actor"% akkaVersion libraryDependencies + = "com.typesafe.akka" %% "akka-testkit"% akkaVersion

Deze hebben we nodig om de Akka Framework-code te kunnen gebruiken en testen.

Vervolgens gaan we Akka-streams gebruiken. Dus laten we de akka-stream afhankelijkheid:

libraryDependencies + = "com.typesafe.akka" %% "akka-stream"% akkaVersion

Ten slotte moeten we een rusteindpunt aanroepen van een Akka-actor. Hiervoor hebben we de akka-http afhankelijkheid. Wanneer we dit doen, retourneert het eindpunt JSON-gegevens die we moeten deserialiseren, dus we moeten de akka-http-jackson afhankelijkheid ook:

libraryDependencies + = "com.typesafe.akka" %% "akka-http-jackson"% akkaHttpVersion libraryDependencies + = "com.typesafe.akka" %% "akka-http"% akkaHttpVersion

En nu zijn we helemaal klaar. Laten we eens kijken hoe we WebSockets kunnen laten werken!

3. Omgaan met WebSockets met Akka-acteurs

Het WebSocket-afhandelingsmechanisme van Play is gebouwd rond Akka-streams. Een WebSocket wordt gemodelleerd als een Flow. Inkomende WebSocket-berichten worden dus in de stroom ingevoerd en berichten die door de stroom worden geproduceerd, worden naar de client verzonden.

Om een ​​WebSocket met een acteur af te handelen, hebben we het hulpprogramma Play nodig ActorFlow die een Actor Ref naar een stroom. Dit vereist voornamelijk wat Java-code, met een kleine configuratie.

3.1. De WebSocket-controllermethode

Ten eerste hebben we een Materializer voorbeeld. De Materializer is een fabriek voor stream-executiemotoren.

We moeten de ActorSystem en de Materializer in de controller app / controllers / HomeController.java:

privé ActorSystem actorSystem; particuliere Materializer materializer; @Inject openbare HomeController (ActorSystem actorSystem, Materializer materializer) {this.actorSystem = actorSystem; this.materializer = materializer; }

Laten we nu een socketcontroller-methode toevoegen:

openbare WebSocket-socket () {retourneer WebSocket.Json .acceptOrResult (dit :: createActorFlow); }

Hier noemen we de functie acceptOrResult die de verzoekheader pakt en een toekomst retourneert. De geretourneerde toekomst is een stroom om de WebSocket-berichten af ​​te handelen.

In plaats daarvan kunnen we het verzoek afwijzen en een afwijzingsresultaat retourneren.

Laten we nu de stroom maken:

privé Voltooiingsfase<>> createActorFlow (verzoek Http.RequestHeader) {retourneer CompletableFuture.completedFuture (F.Either.Right (createFlowForActor ())); }

De F. class in Play Framework definieert een reeks functionele helpers voor programmeerstijlen. In dit geval gebruiken we F.Ofwel, juist om de verbinding te accepteren en de stroom terug te sturen.

Laten we zeggen dat we de verbinding wilden weigeren als de client niet geauthenticeerd is.

Hiervoor kunnen we controleren of er een gebruikersnaam is ingesteld in de sessie. En als dat niet het geval is, weigeren we de verbinding met HTTP 403 Forbidden:

privé Voltooiingsfase<>> createActorFlow2 (Http.RequestHeader-verzoek) {retourneer CompletableFuture.completedFuture (request.session () .getOptional ("gebruikersnaam") .map (gebruikersnaam -> F.Either.Right (createFlowForActor ())) .orElseGet (() -> F.Either.Left (forbidden ()))); }

We gebruiken F. Ofwel Links om de verbinding op dezelfde manier te weigeren als waarmee we een stroom bieden Ofwel, juist.

Ten slotte koppelen we de stroom aan de actor die de berichten zal afhandelen:

private Flow createFlowForActor () {retourneer ActorFlow.actorRef (uit -> Messenger.props (uit), actorSystem, materializer); }

De ActorFlow.actorRef creëert een stroom die wordt afgehandeld door de Boodschapper acteur.

3.2. De routes het dossier

Laten we nu het routes definities voor de controllermethoden in conf / routes:

GET / controllers.HomeController.index (verzoek: Request) GET / chat controllers.HomeController.socket GET / chat / met / streams controllers.HomeController.akkaStreamsSocket GET / assets / * file controllers.Assets.versioned (path = "/ public" , bestand: Asset)

Deze routedefinities wijzen inkomende HTTP-verzoeken toe aan actiemethoden van de controller, zoals uitgelegd in Routing in Play-applicaties in Java.

3.3. De Actor-implementatie

Het belangrijkste onderdeel van de actorklasse is de createReceive methode die bepaalt welke berichten de acteur kan afhandelen:

@Override public Receive createReceive () {return ReceiveBuilder () .match (JsonNode.class, this :: onSendMessage) .matchAny (o -> log.error ("Onbekend bericht ontvangen: {}", o.getClass ())) .bouwen(); }

De acteur stuurt alle berichten die overeenkomen met het JsonNode klasse naar de onSendMessage handler methode:

private void onSendMessage (JsonNode jsonNode) {RequestDTO requestDTO = MessageConverter.jsonNodeToRequest (jsonNode); String bericht = requestDTO.getMessage (). ToLowerCase (); // .. processMessage (requestDTO); }

Vervolgens zal de handler op elk bericht reageren met behulp van de processMessage methode:

private void processMessage (RequestDTO requestDTO) {CompletionStage responseFuture = getRandomMessage (); responseFuture.thenCompose (this :: consumeHttpResponse) .thenAccept (messageDTO -> out.tell (MessageConverter.messageToJsonNode (messageDTO), getSelf ())); }

3.4. Rest API gebruiken met Akka HTTP

We sturen HTTP-verzoeken naar de dummy-berichtgenerator op JSONPlaceholder Posts. Wanneer het antwoord binnenkomt, sturen we het antwoord naar de klant door het te schrijven uit.

Laten we een methode hebben die het eindpunt aanroept met een willekeurige post-ID:

privé CompletionStage getRandomMessage () {int postId = ThreadLocalRandom.current (). nextInt (0, 100); retourneer Http.get (getContext (). getSystem ()) .singleRequest (HttpRequest.create ("//jsonplaceholder.typicode.com/posts/" + postId)); }

We verwerken ook de HttpResponse we krijgen van het aanroepen van de service om het JSON-antwoord te krijgen:

privé CompletionStage consumeHttpResponse (HttpResponse httpResponse) {Materializer materializer = Materializer.matFromSystem (getContext (). getSystem ()); terugkeer Jackson.unmarshaller (MessageDTO.class) .unmarshal (httpResponse.entity (), materializer) .thenApply (messageDTO -> {log.info ("Ontvangen bericht: {}", messageDTO); discardEntity (httpResponse, materializer); terug messageDTO;}); }

De MessageConverter class is een hulpprogramma voor het converteren tussen JsonNode en de DTO's:

openbare statische MessageDTO jsonNodeToMessage (JsonNode jsonNode) {ObjectMapper mapper = nieuwe ObjectMapper (); return mapper.convertValue (jsonNode, MessageDTO.class); }

Vervolgens moeten we de entiteit weggooien. De discardEntityBytes gemaksmethode dient om de entiteit gemakkelijk te verwijderen als deze geen doel voor ons heeft.

Laten we eens kijken hoe we de bytes kunnen verwijderen:

private void discardEntity (HttpResponse httpResponse, Materializer materializer) {HttpMessage.DiscardedEntity discarded = httpResponse.discardEntityBytes (materializer); discarded.completionStage () .whenComplete ((done, ex) -> log.info ("Entiteit volledig weggegooid!")); }

Nu we de afhandeling van de WebSocket hebben gedaan, laten we eens kijken hoe we hiervoor een client kunnen opzetten met HTML5 WebSockets.

4. Het opzetten van de WebSocket Client

Laten we voor onze klant een eenvoudige webgebaseerde chattoepassing bouwen.

4.1. De controlleractie

We moeten een controlleractie definiëren die de indexpagina weergeeft. We plaatsen dit in de controller-klasse app.controllers.HomeController:

openbare resultatenindex (verzoek Http.Request) {String url = routes.HomeController.socket () .webSocketURL (verzoek); retourneer ok (views.html.index.render (url)); } 

4.2. De sjabloonpagina

Laten we nu naar de app / views / ndex.scala.html pagina en voeg een container toe voor de ontvangen berichten en een formulier om een ​​nieuw bericht vast te leggen:

 F Verzenden 

We moeten ook de URL voor de WebSocket-controlleractie doorgeven door deze parameter bovenaan de app / views / index.scala.htmlbladzijde:

@ (url: tekenreeks)

4.3. WebSocket Event Handlers in JavaScript

En nu kunnen we JavaScript toevoegen om de WebSocket-gebeurtenissen af ​​te handelen. Voor de eenvoud voegen we de JavaScript-functies onderaan het app / views / index.scala.html bladzijde.

Laten we de event handlers declareren:

var webSocket; var messageInput; functie init () {initWebSocket (); } functie initWebSocket () {webSocket = nieuwe WebSocket ("@ url"); webSocket.onopen = onOpen; webSocket.onclose = onClose; webSocket.onmessage = onMessage; webSocket.onerror = onError; }

Laten we de handlers zelf toevoegen:

functie onOpen (evt) {writeToScreen ("CONNECTED"); } function onClose (evt) {writeToScreen ("DISCONNECTED"); } function onError (evt) {writeToScreen ("ERROR:" + JSON.stringify (evt)); } functie onMessage (evt) {var ontvangenData = JSON.parse (evt.data); appendMessageToView ("Server", ontvangenData.body); }

Vervolgens gebruiken we de functies om de uitvoer te presenteren appendMessageToView en writeToScreen:

functie appendMessageToView (titel, bericht) {$ ("# messageContent"). append ("
"+ title +": "+ bericht +"
");} functie writeToScreen (bericht) {console.log (" Nieuw bericht: ", bericht);}

4.4. De applicatie uitvoeren en testen

We zijn klaar om de applicatie te testen, dus laten we hem uitvoeren:

cd websockets sbt uitvoeren

Terwijl de applicatie draait, kunnen we met de server chatten door te bezoeken // localhost: 9000:

Elke keer dat we een bericht typen en op Sturen de server zal onmiddellijk reageren met een aantal lorem ipsum van de JSON Placeholder-service.

5. Direct omgaan met WebSockets met Akka Streams

Als we een stroom gebeurtenissen uit een bron verwerken en deze naar de klant sturen, dan kunnen we dit rond Akka-streams modelleren.

Laten we eens kijken hoe we Akka-streams kunnen gebruiken in een voorbeeld waarbij de server elke twee seconden berichten verzendt.

We beginnen met de WebSocket-actie in het HomeController:

openbare WebSocket akkaStreamsSocket () {retourneer WebSocket.Json.accept (request -> {Sink in = Sink.foreach (System.out :: println); MessageDTO messageDTO = new MessageDTO ("1", "1", "Title", "Test Body"); Source out = Source.tick (Duration.ofSeconds (2), Duration.ofSeconds (2), MessageConverter.messageToJsonNode (messageDTO)); return Flow.fromSinkAndSource (in, out);}); }

De Bron#Kruis aan methode heeft drie parameters. De eerste is de aanvankelijke vertraging voordat de eerste tik wordt verwerkt, en de tweede is het interval tussen opeenvolgende tikjes. We hebben beide waarden ingesteld op twee seconden in het bovenstaande fragment. De derde parameter is een object dat bij elke tik moet worden geretourneerd.

Om dit in actie te zien, moeten we de URL in het inhoudsopgave actie en laat het verwijzen naar de akkaStreamsSocket eindpunt:

String url = routes.HomeController.akkaStreamsSocket (). WebSocketURL (verzoek);

En nu de pagina wordt vernieuwd, zien we elke twee seconden een nieuw item:

6. Beëindiging van de acteur

Op een gegeven moment zullen we de chat moeten afsluiten, hetzij via een gebruikersverzoek, hetzij via een time-out.

6.1. Afhandelen van de beëindiging van de acteur

Hoe detecteren we wanneer een WebSocket is gesloten?

Het afspelen zal automatisch de WebSocket sluiten wanneer de actor die de WebSocket afhandelt stopt. We kunnen dit scenario dus aan door het Acteur # postStop methode:

@Override public void postStop () gooit Uitzondering {log.info ("Messenger-actor gestopt om {}", OffsetDateTime.now () .format (DateTimeFormatter.ISO_OFFSET_DATE_TIME)); }

6.2. De acteur handmatig beëindigen

Verder, als we de acteur moeten stoppen, kunnen we een Vergifpil aan de acteur. In onze voorbeeldtoepassing zouden we een "stop" -verzoek moeten kunnen afhandelen.

Laten we eens kijken hoe we dit moeten doen in het onSendMessage methode:

private void onSendMessage (JsonNode jsonNode) {RequestDTO requestDTO = MessageConverter.jsonNodeToRequest (jsonNode); String bericht = requestDTO.getMessage (). ToLowerCase (); if ("stop" .equals (bericht)) {MessageDTO messageDTO = createMessageDTO ("1", "1", "Stop", "Stopping actor"); out.tell (MessageConverter.messageToJsonNode (messageDTO), getSelf ()); self (). tell (PoisonPill.getInstance (), getSelf ()); } else {log.info ("Actor ontvangen. {}", requestDTO); processMessage (requestDTO); }}

Als we een bericht ontvangen, kijken we of het een stopverzoek is. Als dit het geval is, sturen we het Vergifpil. Anders verwerken we het verzoek.

7. Configuratie-opties

We kunnen verschillende opties configureren in termen van hoe de WebSocket moet worden behandeld. Laten we er een paar bekijken.

7.1. Lengte WebSocket-frame

WebSocket-communicatie omvat de uitwisseling van dataframes.

De lengte van het WebSocket-frame is configureerbaar. We hebben de mogelijkheid om de framelengte aan te passen aan onze toepassingsvereisten.

Het configureren van een kortere framelengte kan helpen bij het verminderen van Denial of Service-aanvallen waarbij lange dataframes worden gebruikt. We kunnen de framelengte voor de toepassing wijzigen door de maximale lengte op te geven in application.conf:

play.server.websocket.frame.maxLength = 64k

We kunnen deze configuratieoptie ook instellen door de maximale lengte op te geven als een opdrachtregelparameter:

sbt -Dwebsocket.frame.maxLength = 64k uitvoeren

7.2. Time-out bij inactiviteit van de verbinding

Standaard wordt de actor die we gebruiken om de WebSocket af te handelen na één minuut beëindigd. Dit komt doordat de Play-server waarop onze applicatie wordt uitgevoerd, een standaard time-out voor inactiviteit van 60 seconden heeft. Dit betekent dat alle verbindingen die binnen zestig seconden geen verzoek ontvangen, automatisch worden gesloten.

We kunnen dit veranderen door middel van configuratie-opties. Laten we naar onze gaan application.conf en verander de server om geen time-out voor inactiviteit te hebben:

play.server.http.idleTimeout = "oneindig"

Of we kunnen de optie doorgeven als opdrachtregelargumenten:

sbt -Dhttp.idleTimeout = oneindige run

We kunnen dit ook configureren door te specificeren devSettings in build.sbt.

Configuratie-opties gespecificeerd in build.sbt worden alleen gebruikt bij de ontwikkeling, ze worden genegeerd bij de productie:

PlayKeys.devSettings + = "play.server.http.idleTimeout" -> "oneindig"

Als we de applicatie opnieuw starten, wordt de actor niet beëindigd.

We kunnen de waarde veranderen in seconden:

PlayKeys.devSettings + = "play.server.http.idleTimeout" -> "120 s"

We kunnen meer lezen over de beschikbare configuratie-opties in de Play Framework-documentatie.

8. Conclusie

In deze tutorial hebben we WebSockets geïmplementeerd in het Play Framework met Akka-acteurs en Akka Streams.

We gingen vervolgens kijken hoe we Akka-acteurs rechtstreeks konden gebruiken en zagen vervolgens hoe Akka Streams kunnen worden ingesteld om de WebSocket-verbinding af te handelen.

Aan de clientzijde hebben we JavaScript gebruikt om onze WebSocket-evenementen af ​​te handelen.

Ten slotte hebben we enkele configuratie-opties bekeken die we kunnen gebruiken.

Zoals gewoonlijk is de broncode voor deze tutorial beschikbaar op GitHub.