streamalism.org

1. Oversikt

Når vi ønsker at våre webklienter skal opprettholde en dialog med serveren vår, kan WebSockets være en nyttig løsning. WebSockets holder en vedvarende full-dupleks-tilkobling. Dette gir oss muligheten til å sende toveis meldinger mellom serveren og klienten.

I denne opplæringen skal vi lære hvordan du bruker WebSockets med Akka i Play Framework.

2. Oppsett

La oss sette opp en enkel chat-applikasjon. Brukeren vil sende meldinger til serveren, og serveren vil svare med en melding fra JSONPlaceholder.

2.1. Sette opp Play Framework-applikasjonen

Vi bygger dette programmet ved hjelp av Play Framework.

La oss følge instruksjonene fra Introduksjon til Play i Java for å sette opp og kjøre et enkelt Play Framework-program.

2.2. Legge til nødvendige JavaScript-filer

Vi må også jobbe med JavaScript for skripting på klientsiden. Dette vil gjøre det mulig for oss å motta nye meldinger presset fra serveren. Vi bruker jQuery-biblioteket til dette.

La oss legge til jQuery i bunnen av app / visninger / index.scala.html fil:

2.3. Sette opp Akka

Til slutt bruker vi Akka til å håndtere WebSocket-tilkoblingene på serversiden.

La oss navigere til build.sbt fil og legg til avhengighetene.

Vi må legge til akka-skuespiller og akka-testkit avhengigheter:

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

Vi trenger disse for å kunne bruke og teste Akka Framework-koden.

Deretter skal vi bruke Akka-strømmer. Så la oss legge til akka-stream avhengighet:

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

Til slutt må vi kalle et hvilepunkt fra en Akka-skuespiller. For dette trenger vi akka-http avhengighet. Når vi gjør det, vil endepunktet returnere JSON-data som vi må deserialisere, så vi må legge til akka-http-jackson avhengighet også:

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

Og nå er vi klare. La oss se hvordan vi får WebSockets til å fungere!

3. Håndtering av WebSockets med Akka Actors

Play's WebSocket-håndteringsmekanisme er bygget rundt Akka-strømmer. En WebSocket er modellert som en Flow. Så innkommende WebSocket-meldinger mates inn i flyten, og meldinger produsert av strømmen sendes ut til klienten.

For å håndtere en WebSocket ved hjelp av en skuespiller, trenger vi Play-verktøyet SkuespillerFlow som konverterer en SkuespillerRef til en flyt. Dette krever hovedsakelig litt Java-kode, med litt konfigurasjon.

3.1. WebSocket Controller-metoden

Først trenger vi en Materializer forekomst. Materializer er en fabrikk for strømkjøringsmotorer.

Vi må injisere ActorSystem og Materializer inn i kontrolleren app / kontrollere / HomeController.java:

private ActorSystem actorSystem; private materialiseringsmaterialer; @Injiser offentlig HomeController (ActorSystem actorSystem, Materializer materializer) {this.actorSystem = actorSystem; this.materializer = materializer; }

La oss nå legge til en kontaktkontrollmetode:

offentlig WebSocket-kontakt () {return WebSocket.Json .acceptOrResult (dette :: createActorFlow); }

Her kaller vi funksjonen godtaOrResult som tar forespørselstittelen og returnerer en fremtid. Den returnerte fremtiden er en strøm for å håndtere WebSocket-meldingene.

Vi kan i stedet avvise forespørselen og returnere et avvisningsresultat.

La oss nå lage strømmen:

private CompletionStage<>> createActorFlow (Http.RequestHeader-forespørsel) {return CompletableFuture.completedFuture (F.Either.Right (createFlowForActor ())); }

De F klasse i Play Framework definerer et sett med funksjonelle hjelpere for programmeringsstil. I dette tilfellet bruker vi F.Enten. Rett for å godta tilkoblingen og returnere flyten.

La oss si at vi ønsket å avvise forbindelsen når klienten ikke er godkjent.

For dette kan vi sjekke om et brukernavn er angitt i økten. Og hvis det ikke er det, avviser vi forbindelsen med HTTP 403 Forbidden:

private CompletionStage<>> createActorFlow2 (Http.RequestHeader-forespørsel) {retur CompletableFuture.completedFuture (request.session () .getOptional ("brukernavn"). kart (brukernavn -> F. Enten.Høyre (createFlowForActor ())). EllerElseGet (() -> F.Either.Left (forbudt ()))); }

Vi bruker F.Enten. Venstre å avvise forbindelsen på samme måte som vi gir en flyt med F. Enten. Rett.

Til slutt kobler vi strømmen til skuespilleren som skal håndtere meldingene:

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

De ActorFlow.actorRef skaper en flyt som håndteres av budbringer skuespiller.

3.2. De ruter Fil

La oss nå legge til ruter definisjoner for kontrollermetodene i conf / ruter:

GET / controllers.HomeController.index (forespørsel: Request) GET / chat controllers.HomeController.socket GET / chat / with / streams controllers.HomeController.akkaStreamsSocket GET / assets / * file controllers.Assets.versioned (path = "/ public" , fil: aktiva)

Disse rutedefinisjonene kartlegger innkommende HTTP-forespørsler til handlingsmetoder for kontrolleren som forklart i Routing i Play Applications i Java.

3.3. Skuespillerimplementeringen

Den viktigste delen av skuespillerklassen er createReceive metode som bestemmer hvilke meldinger skuespilleren kan håndtere:

@ Override public Receive createReceive () {return receiveBuilder () .match (JsonNode.class, this :: onSendMessage) .matchAny (o -> log.error ("Mottatt ukjent melding: {}", o.getClass ())) .bygge(); }

Skuespilleren vil videresende alle meldinger som samsvarer med JsonNode klasse til onSendMessage behandler metode:

privat tomrom onSendMessage (JsonNode jsonNode) {RequestDTO requestDTO = MessageConverter.jsonNodeToRequest (jsonNode); Strengmelding = requestDTO.getMessage (). ToLowerCase (); // .. processMessage (requestDTO); }

Da vil behandleren svare på alle meldinger ved hjelp av processMessage metode:

privat ugyldig prosessMessage (RequestDTO requestDTO) {CompletionStage responseFuture = getRandomMessage (); responseFuture.thenCompose (dette :: consumeHttpResponse) .thenAccept (messageDTO -> out.tell (MessageConverter.messageToJsonNode (messageDTO), getSelf ())); }

3.4. Forbruker Rest API med Akka HTTP

Vi sender HTTP-forespørsler til dummy-meldingsgeneratoren på JSONPlaceholder Posts. Når svaret kommer, sender vi svaret til klienten ved å skrive det ute.

La oss ha en metode som kaller sluttpunktet med en tilfeldig post-ID:

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

Vi behandler også HttpResponse vi får fra å ringe tjenesten for å få JSON-svaret:

private CompletionStage consumeHttpResponse (HttpResponse httpResponse) {Materializer materializer = Materializer.matFromSystem (getContext (). getSystem ()); returner Jackson.unmarshaller (MessageDTO.class) .unmarshal (httpResponse.entity (), materializer) .thenApply (messageDTO -> {log.info ("Mottatt melding: {}", messageDTO); discardEntity (httpResponse, materializer); retur messageDTO;}); }

De MessageConverter class er et verktøy for konvertering mellom JsonNode og DTO-ene:

offentlig statisk MessageDTO jsonNodeToMessage (JsonNode jsonNode) {ObjectMapper mapper = new ObjectMapper (); returner mapper.convertValue (jsonNode, MessageDTO.class); }

Deretter må vi forkaste enheten. De kastEntityBytes bekvemmelighetsmetode tjener formålet med å kaste enheten enkelt hvis den ikke har noe formål for oss.

La oss se hvordan vi kan forkaste byte:

private void discardEntity (HttpResponse httpResponse, Materializer materializer) {HttpMessage.DiscardedEntity forkastet = httpResponse.discardEntityBytes (materializer); discarded.completionStage () .whenComplete ((gjort, ex) -> log.info ("Enheten forkastet fullstendig!"); }

Etter å ha gjort håndteringen av WebSocket, la oss se hvordan vi kan sette opp en klient for dette ved hjelp av HTML5 WebSockets.

4. Sette opp WebSocket Client

For vår klient, la oss bygge en enkel nettbasert chat-applikasjon.

4.1. Kontrollerhandlingen

Vi må definere en kontrollerhandling som gjengir indeksiden. Vi legger dette i kontrollerklassen app.controllers.HomeController:

offentlig resultatindeks (Http.Request forespørsel) {String url = routes.HomeController.socket () .webSocketURL (forespørsel); returner ok (views.html.index.render (url)); } 

4.2. Mal-siden

La oss gå over til app / views / ndex.scala.html side og legg til en container for de mottatte meldingene og et skjema for å fange en ny melding:

 F Send 

Vi må også sende inn URL-en for handlingen WebSocket-kontroller ved å erklære denne parameteren øverst på app / views / index.scala.htmlside:

@ (url: streng)

4.3. WebSocket Event Handlers i JavaScript

Og nå kan vi legge til JavaScript for å håndtere WebSocket-hendelsene. For enkelhets skyld vil vi legge til JavaScript-funksjonene nederst på app / views / index.scala.html side.

La oss erklære hendelsesbehandlerne:

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

La oss legge til håndtererne selv:

funksjon onOpen (evt) {writeToScreen ("CONNECTED"); } funksjon onClose (evt) {writeToScreen ("FRAKOBLET"); } funksjon onError (evt) {writeToScreen ("FEIL:" + JSON.stringify (evt)); } funksjon onMessage (evt) {var receivedData = JSON.parse (evt.data); appendMessageToView ("Server", receivedData.body); }

For å presentere utdataene bruker vi funksjonene appendMessageToView og skriv til skjerm:

funksjon appendMessageToView (tittel, melding) {$ ("# messageContent"). append ("
"+ title +": "+ melding +"
");} funksjon writeToScreen (melding) {console.log (" Ny melding: ", melding);}

4.4. Kjøre og teste applikasjonen

Vi er klare til å teste applikasjonen, så la oss kjøre den:

cd websockets sbt run

Når applikasjonen kjører, kan vi chatte med serveren ved å besøke // lokal vert: 9000:

Hver gang vi skriver inn en melding og treffer Sende serveren vil umiddelbart svare med noen lorem ipsum fra JSON Placeholder-tjenesten.

5. Håndtere WebSockets direkte med Akka Streams

Hvis vi behandler en strøm av hendelser fra en kilde og sender disse til klienten, kan vi modellere dette rundt Akka-strømmer.

La oss se hvordan vi kan bruke Akka-strømmer i et eksempel der serveren sender meldinger hvert annet sekund.

Vi starter med WebSocket-handlingen i HomeController:

offentlig WebSocket akkaStreamsSocket () {return WebSocket.Json.accept (forespørsel -> {Sink in = Sink.foreach (System.out :: println); MessageDTO messageDTO = new MessageDTO ("1", "1", "Tittel", "Test Body"); Source out = Source.tick (Duration.ofSeconds (2), Duration.ofSeconds (2), MessageConverter.messageToJsonNode (messageDTO)); returner Flow.fromSinkAndSource (inn, ut);}); }

De Kilde#sett kryss metoden tar tre parametere. Den første er den innledende forsinkelsen før det første krysset behandles, og den andre er intervallet mellom påfølgende flått. Vi har satt begge verdiene til to sekunder i kodebiten ovenfor. Den tredje parameteren er et objekt som skal returneres ved hvert kryss.

For å se dette i aksjon, må vi endre URL-adressen i indeks handling og få den til å peke på akkaStreamsSocket sluttpunkt:

String url = routes.HomeController.akkaStreamsSocket (). WebSocketURL (forespørsel);

Og nå oppdaterer vi siden, vi ser en ny oppføring hvert annet sekund:

6. Avslutte skuespilleren

På et tidspunkt må vi slå av chatten, enten gjennom en brukerforespørsel eller gjennom en tidsavbrudd.

6.1. Håndtering av skuespilleroppsigelse

Hvordan oppdager vi når en WebSocket er stengt?

Spill vil automatisk lukke WebSocket når skuespilleren som håndterer WebSocket avsluttes. Så vi kan håndtere dette scenariet ved å implementere Skuespiller # postStop metode:

@ Overstyr offentlig ugyldig postStop () kaster unntak {log.info ("Messenger-skuespiller stoppet kl. {}", OffsetDateTime.now () .format (DateTimeFormatter.ISO_OFFSET_DATE_TIME)); }

6.2. Manuell avslutning av skuespilleren

Videre, hvis vi må stoppe skuespilleren, kan vi sende en PoisonPill til skuespilleren. I vårt eksempel på applikasjon skal vi kunne håndtere en "stopp" -forespørsel.

La oss se hvordan du gjør dette i onSendMessage metode:

privat tomrom onSendMessage (JsonNode jsonNode) {RequestDTO requestDTO = MessageConverter.jsonNodeToRequest (jsonNode); Strengmelding = requestDTO.getMessage (). ToLowerCase (); if ("stop" .equals (message)) {MessageDTO messageDTO = createMessageDTO ("1", "1", "Stop", "Stopping actor"); out.tell (MessageConverter.messageToJsonNode (messageDTO), getSelf ()); self (). tell (PoisonPill.getInstance (), getSelf ()); } annet {log.info ("Skuespiller mottatt. {}", requestDTO); processMessage (requestDTO); }}

Når vi mottar en melding, sjekker vi om det er en stoppforespørsel. Hvis det er det, sender vi PoisonPill. Ellers behandler vi forespørselen.

7. Konfigurasjonsalternativer

Vi kan konfigurere flere alternativer når det gjelder hvordan WebSocket skal håndteres. La oss se på noen få.

7.1. WebSocket rammelengde

WebSocket-kommunikasjon innebærer utveksling av datarammer.

WebSocket-rammelengden kan konfigureres. Vi har muligheten til å justere rammelengden til våre applikasjonskrav.

Konfigurering av kortere rammelengde kan bidra til å redusere denial of service-angrep som bruker lange datarammer. Vi kan endre rammelengden for applikasjonen ved å spesifisere maks lengde i application.conf:

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

Vi kan også angi dette konfigurasjonsalternativet ved å spesifisere maks lengde som en kommandolinjeparameter:

sbt -Dwebsocket.frame.maxLength = 64k kjøring

7.2. Tidsavbrudd for ledig tilkobling

Som standard avsluttes skuespilleren vi bruker for å håndtere WebSocket etter ett minutt. Dette er fordi Play-serveren der applikasjonen vår kjører har en standard tidsavbrudd for inaktivitet på 60 sekunder. Dette betyr at alle forbindelser som ikke mottar en forespørsel på seksti sekunder, lukkes automatisk.

Vi kan endre dette gjennom konfigurasjonsalternativer. La oss gå over til vår application.conf og endre serveren slik at den ikke har ledig tidsavbrudd:

play.server.http.idleTimeout = "uendelig"

Eller vi kan gi inn alternativet som kommandolinjeargumenter:

sbt -Dhttp.idleTimeout = uendelig løp

Vi kan også konfigurere dette ved å spesifisere devSettings i build.sbt.

Konfigurasjonsalternativer spesifisert i build.sbt brukes bare i utvikling, vil de bli ignorert i produksjonen:

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

Hvis vi kjører applikasjonen på nytt, avslutter ikke skuespilleren.

Vi kan endre verdien til sekunder:

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

Vi kan finne ut mer om de tilgjengelige konfigurasjonsalternativene i Play Framework-dokumentasjonen.

8. Konklusjon

I denne veiledningen implementerte vi WebSockets i Play Framework med Akka-skuespillere og Akka Streams.

Vi fortsatte med å se på hvordan du bruker Akka-skuespillere direkte, og så hvordan Akka Streams kan konfigureres for å håndtere WebSocket-tilkoblingen.

På klientsiden brukte vi JavaScript til å håndtere våre WebSocket-hendelser.

Til slutt så vi på noen konfigurasjonsalternativer som vi kan bruke.

Som vanlig er kildekoden for denne opplæringen tilgjengelig på GitHub.