streamalism.org

1. סקירה כללית

כאשר אנו מאחלים ללקוחות האינטרנט שלנו לקיים דיאלוג עם השרת שלנו, אז WebSockets יכול להיות פיתרון שימושי. WebSockets שומרים על חיבור מתמשך של דופלקס מלא. זֶה נותן לנו את היכולת לשלוח הודעות דו כיווניות בין השרת ללקוח שלנו.

במדריך זה נלמד כיצד להשתמש ב- WebSockets עם Akka במסגרת Play Play.

2. התקנה

בואו נקבע אפליקציית צ'אט פשוטה. המשתמש ישלח הודעות לשרת, והשרת יגיב בהודעה מ- JSONPlaceholder.

2.1. הגדרת יישום Play Framework

אנו בונים יישום זה באמצעות Play Framework.

בואו לעקוב אחר ההוראות ממבוא ל- Play ב- Java כדי להגדיר ולהפעיל יישום Play Framework פשוט.

2.2. הוספת קבצי ה- JavaScript הדרושים

כמו כן, נצטרך לעבוד עם JavaScript לצורך סקריפטים בצד הלקוח. זה יאפשר לנו לקבל הודעות חדשות שנדחקו מהשרת. נשתמש בספריית jQuery לשם כך.

בואו נוסיף את jQuery לתחתית ה- אפליקציה / תצוגות / index.scala.html קוֹבֶץ:

2.3. מקימים עכה

לבסוף נשתמש ב- Akka לטיפול בחיבורי WebSocket בצד השרת.

בוא ננווט אל ה- build.sbt הקובץ והוסף את התלות.

עלינו להוסיף את אקא-שחקן ו akka-testkit תלות:

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

אנו זקוקים לאלו כדי שנוכל להשתמש בקוד Akka Framework ולבחון אותו.

בשלב הבא אנו נשתמש בזרמי עכו. אז בואו נוסיף את akka-stream תלות:

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

לבסוף, עלינו לקרוא לנקודת סיום מנוחה משחקן עכו. לשם כך נצטרך את akka-http תלות. כאשר אנו עושים זאת, נקודת הקצה תחזיר את נתוני JSON אשר נצטרך לנתק מחדש, לכן עלינו להוסיף את akka-http-jackson תלות גם כן:

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

ועכשיו הכל מסודר. בואו נראה איך לגרום ל- WebSockets לעבוד!

3. טיפול בשקעי רשת עם שחקני Akka

מנגנון הטיפול ב- WebSocket של פליי בנוי סביב זרמי עכו. WebSocket מעוצב כזרם. לכן, הודעות WebSocket נכנסות מוזרמות לזרם והודעות שהופקו על ידי הזרימה נשלחות ללקוח.

כדי לטפל ב- WebSocket באמצעות שחקן, נצטרך את תוכנית השירות Play ActorFlow הממיר שחקן רף לזרם. זה בעיקר דורש קצת קוד Java, עם מעט תצורה.

3.1. שיטת בקר ה- WebSocket

ראשית, אנחנו צריכים a מטריאליזציה למשל. ה- Materializer הוא מפעל למנועי ביצוע זרמים.

אנחנו צריכים להזריק את מערכת שחקנים וה מטריאליזציה לתוך הבקר אפליקציה / בקרים / HomeController.java:

שחקן פרטי שחקן מערכת שחקן מערכת; חומר ממטריזר פרטי; @Inject הציבורית HomeController (ActorSystem actorSystem, Materializer Materializer) {this.actorSystem = actorSystem; this.materializer = materializer; }

בואו כעת נוסיף שיטת בקר שקע:

שקע WebSocket ציבורי () {להחזיר WebSocket.Json .acceptOrResult (זה :: createActorFlow); }

כאן אנו קוראים לפונקציה acceptOrResult שלוקח את כותרת הבקשה ומחזיר עתיד. העתיד שהוחזר הוא זרימה לטיפול בהודעות WebSocket.

נוכל במקום זאת לדחות את הבקשה ולהחזיר תוצאת דחייה.

עכשיו, בואו ניצור את הזרימה:

שלב השלמה פרטי<>> createActorFlow (Http.RequestHeader בקשה) {return CompletableFuture.completedFuture (F.Either.Right (createFlowForActor ())); }

ה F class ב- Play Framework מגדיר קבוצה של עוזרים בסגנון תכנות פונקציונלי. במקרה זה, אנו משתמשים פ.או. נכון לקבל את החיבור ולהחזיר את הזרימה.

נניח שרצינו לדחות את החיבור כאשר הלקוח אינו מאומת.

לשם כך נוכל לבדוק אם מוגדר שם משתמש בפגישה. ואם לא, אנו דוחים את החיבור עם HTTP 403 אסור:

שלב השלמה פרטי<>> createActorFlow2 (Http.RequestHeader request) {return CompletableFuture.completedFuture (request.session () .getOptional ("שם משתמש") .map (שם משתמש -> F.Either.מימין (createFlowForActor ())). OrElseGet (() -> F.Either.Left (אסור ()))); }

אנו משתמשים F.Either. שמאל לדחות את החיבור באותו אופן בו אנו מספקים זרימה F. או. נכון.

לבסוף, אנו קושרים את הזרימה לשחקן שיטפל בהודעות:

זרימה פרטית createFlowForActor () {return ActorFlow.actorRef (out -> Messenger.props (out), actorSystem, materializer); }

ה ActorFlow.actorRef יוצר זרימה שמטופלת על ידי שָׁלִיחַ שַׂחְקָן.

3.2. ה מסלולים קוֹבֶץ

עכשיו, בואו נוסיף את ה- מסלולים הגדרות לשיטות הבקר ב conf / מסלולים:

GET / controllers.HomeController.index (בקשה: בקשה) GET / controllers צ'אט. HomeController.socket GET / chat / with / streams controllers.HomeController.akkaStreamsSocket GET / נכסים / * בקרי קבצים. Assets.versioned (path = "/ public" , קובץ: נכס)

הגדרות מסלול אלה ממפות בקשות HTTP נכנסות לשיטות פעולה של בקר, כפי שהוסבר בניתוב ביישומי Play ב- Java.

3.3. יישום השחקן

החלק החשוב ביותר בשיעור השחקנים הוא ה- createReceive שיטה שקובע אילו מסרים השחקן יכול לטפל:

@Override ציבור קבל createReceive () {return receiveBuilder () .match (JsonNode.class, זה :: onSendMessage) .matchAny (o -> log.error ("התקבלה הודעה לא ידועה: {}", o.getClass ())) .לִבנוֹת(); }

השחקן יעביר את כל ההודעות התואמות את JsonNode כיתה ל onSendMessage שיטת המטפל:

חלל פרטי onSendMessage (JsonNode jsonNode) {RequestDTO requestDTO = MessageConverter.jsonNodeToRequest (jsonNode); הודעת מחרוזת = requestDTO.getMessage (). ToLowerCase (); // .. processMessage (requestDTO); }

ואז המטפל יגיב לכל הודעה באמצעות ה- processMessage שיטה:

תהליך ריק בטלפוןMessage (RequestDTO requestDTO) {CompletionStage responseFuture = getRandomMessage (); responseFuture.thenCompose (זה :: consumeHttpResponse) .thenAccept (messageDTO -> out.tell (MessageConverter.messageToJsonNode (messageDTO), getSelf ())); }

3.4. צורכת Rest API עם Akka HTTP

אנו נשלח בקשות HTTP למחולל הודעות הדמה ב- JSONPlaceholder Posts. כאשר התגובה מגיעה, אנו שולחים את התגובה ללקוח בכתיבתה הַחוּצָה.

נקבל שיטה שקוראת לנקודת הקצה עם מזהה פוסט אקראי:

CompletionStage פרטי getRandomMessage () {int postId = ThreadLocalRandom.current (). nextInt (0, 100); להחזיר Http.get (getContext (). getSystem ()). singleRequest (HttpRequest.create ("//jsonplaceholder.typicode.com/posts/" + postId)); }

אנו גם מעבדים את HttpResponse אנו מקבלים משיחת שירות על מנת לקבל את תגובת JSON:

השלמה פרטית Stage consumumeHttpResponse (HttpResponse httpResponse) {Materializer materializer = Materializer.matFromSystem (getContext (). getSystem ()); החזר את Jackson.unmarshaller (MessageDTO.class) .unmarshal (httpResponse.entity (), materializer). ואז החל (messageDTO -> {log.info ("הודעה שהתקבלה: {}", messageDTO); discardEntity (httpResponse, materializer); החזר messageDTO;}); }

ה MessageConverter class הוא כלי עזר להמרה בין JsonNode וה- DTO:

MessageDTO jsonNodeToMessage (JsonNode jsonNode) {ממפה ObjectMapper = ObjectMapper חדש (); להחזיר mapper.convertValue (jsonNode, MessageDTO.class); }

לאחר מכן עלינו להשליך את היישות. ה מחק את האנטי-בייט שיטת נוחות משרתת את המטרה להשליך בקלות את היישות אם אין לה מטרה בשבילנו.

בואו נראה כיצד למחוק את הבתים:

בטל פרטי ריק (DiscardEntity) (HttpResponse httpResponse, Materializer Materializer) {HttpMessage.DiscardedEntity נמחק = httpResponse.discardEntityBytes (materializer); discarded.completionStage () .whenComplete ((נעשה, לשעבר) -> log.info ("הישות נמחקה לחלוטין!")); }

לאחר שביצענו את הטיפול ב- WebSocket, בואו נראה כיצד נוכל להגדיר לקוח לכך באמצעות HTML5 WebSockets.

4. הגדרת לקוח WebSocket

עבור הלקוח שלנו, בואו נבנה אפליקציית צ'אט מבוססת אינטרנט פשוטה.

4.1. פעולת הבקר

עלינו להגדיר פעולת בקר שמעבירה את דף האינדקס. נכניס את זה למחלקת הבקר app.controllers.HomeController:

אינדקס תוצאות ציבורי (Http.Request בקשה) {String url = routes.HomeController.socket () .webSocketURL (בקשה); החזר בסדר (views.html.index.render (url)); } 

4.2. דף התבנית

עכשיו, בואו ניגש אל ה- אפליקציה / תצוגות / ndex.scala.html עמוד והוסף מיכל להודעות שהתקבלו וטופס ללכידת הודעה חדשה:

 F שלח 

נצטרך גם להעביר את כתובת ה- URL לפעולת הבקר של WebSocket על ידי הכרזת פרמטר זה בחלק העליון של ה- app / views / index.scala.htmlעמוד:

@ (url: String)

4.3. מטפלי אירועים WebSocket ב- JavaScript

ועכשיו, אנו יכולים להוסיף את JavaScript לטיפול באירועי WebSocket. לשם פשטות, נוסיף את פונקציות ה- JavaScript בתחתית ה- app / views / index.scala.html עמוד.

בואו נכריז על מטפלי האירועים:

var webSocket; var messageInput; function init () {initWebSocket (); } פונקציה initWebSocket () {webSocket = WebSocket חדש ("@ url"); webSocket.onopen = onOpen; webSocket.onclose = onClose; webSocket.onmessage = onMessage; webSocket.onerror = onError; }

בואו נוסיף את המטפלים עצמם:

פונקציה onOpen (evt) {writeToScreen ("מחובר"); } פונקציה onClose (evt) {writeToScreen ("לא מנותק"); } פונקציה onError (evt) {writeToScreen ("שגיאה:" + JSON.stringify (evt)); } פונקציה onMessage (evt) {var receivedData = JSON.parse (evt.data); appendMessageToView ("שרת", receivedData.body); }

ואז, כדי להציג את הפלט, נשתמש בפונקציות appendMessageToView ו writeToScreen:

function appendMessageToView (title, message) {$ ("# messageContent"). append ("
"+ title +": "+ הודעה +"
");} פונקציה writeToScreen (הודעה) {console.log (" הודעה חדשה: ", הודעה);}

4.4. הפעלה ובדיקת היישום

אנחנו מוכנים לבדוק את היישום, אז בואו נפעיל אותו:

תקליטורי רשת CD

כשהאפליקציה פועלת, אנו יכולים לשוחח בצ'אט עם השרת על ידי ביקור // localhost: 9000:

בכל פעם שאנחנו מקלידים הודעה ופוגעים לִשְׁלוֹחַ השרת יגיב מיד עם כמה לורם איפסום משירות JSON Placeholder.

5. טיפול ישירות ב- WebSockets עם זרמי Akka

אם אנו מעבדים זרם אירועים ממקור ושולחים אותם ללקוח, נוכל לדגם זאת סביב זרמי Akka.

בואו נראה כיצד נוכל להשתמש בזרמי Akka בדוגמה שבה השרת שולח הודעות אחת לשתי שניות.

נתחיל בפעולת WebSocket ב HomeController:

WebSocket ציבורי akkaStreamsSocket () {return WebSocket.Json.accept (בקשה -> {Sink in = Sink.foreach (System.out :: println); MessageDTO messageDTO = new MessageDTO ("1", "1", "כותרת", "מבחן גוף"); מקור החוצה = מקור.טיק (Duration.ofSeconds (2), Duration.ofSeconds (2), MessageConverter.messageToJsonNode (messageDTO)); להחזיר Flow.fromSinkAndSource (פנימה, החוצה);}); }

ה מָקוֹר#תִקתוּק השיטה לוקחת שלושה פרמטרים. הראשון הוא העיכוב הראשוני לפני עיבוד הסימון הראשון, והשני הוא המרווח בין קרציות עוקבות. הגדרנו את שני הערכים לשתי שניות בקטע הקוד. הפרמטר השלישי הוא אובייקט שיש להחזיר בכל סימן.

כדי לראות זאת בפעולה, עלינו לשנות את כתובת האתר ב- אינדקס פעולה ולגרום לה להצביע על akkaStreamsSocket נקודת סיום:

מחרוזת url = routes.HomeController.akkaStreamsSocket (). WebSocketURL (בקשה);

ועכשיו מרעננים את הדף, נראה ערך חדש כל שתי שניות:

6. סיום השחקן

בשלב מסוים, נצטרך לסגור את הצ'אט, דרך בקשת משתמש או באמצעות פסק זמן.

6.1. טיפול בסיום שחקן

כיצד נזהה מתי נסגר WebSocket?

המשחק יסגור את ה- WebSocket באופן אוטומטי כאשר השחקן המטפל ב- WebSocket יסתיים. כדי שנוכל להתמודד עם תרחיש זה על ידי יישום ה- שחקן # postStop שיטה:

@ ביטול ציבורי ריק בטל postStop () זורק חריג {log.info ("שחקן המסנג'ר נעצר בשעה {}", OffsetDateTime.now (). פורמט (DateTimeFormatter.ISO_OFFSET_DATE_TIME)); }

6.2. סיום ידני של השחקן

יתר על כן, אם עלינו לעצור את השחקן, נוכל לשלוח גלולת רעל לשחקן. ביישום לדוגמא שלנו, אנו אמורים להיות מסוגלים לטפל בבקשת "עצירה".

בואו נראה איך לעשות זאת ב- onSendMessage שיטה:

חלל פרטי onSendMessage (JsonNode jsonNode) {RequestDTO requestDTO = MessageConverter.jsonNodeToRequest (jsonNode); הודעת מחרוזת = requestDTO.getMessage (). ToLowerCase (); if ("stop" .equals (message)) {MessageDTO messageDTO = createMessageDTO ("1", "1", "Stop", "עצירת שחקן"); out.tell (MessageConverter.messageToJsonNode (messageDTO), getSelf ()); self (). tell (PoisonPill.getInstance (), getSelf ()); } אחר {log.info ("השחקן קיבל. {}", requestDTO); processMessage (requestDTO); }}

כשאנחנו מקבלים הודעה, אנחנו בודקים אם זו בקשת עצירה. אם זה, אנו שולחים את גלולת רעל. אחרת, אנו מעבדים את הבקשה.

7. אפשרויות תצורה

אנו יכולים להגדיר מספר אפשרויות במונחים של אופן הטיפול ב- WebSocket. בואו נסתכל על כמה.

7.1. אורך מסגרת WebSocket

תקשורת WebSocket כוללת החלפה של מסגרות נתונים.

ניתן להגדיר את אורך המסגרת של WebSocket. יש לנו אפשרות להתאים את אורך המסגרת לדרישות היישום שלנו.

קביעת תצורה של אורך פריימים קצר יותר עשויה לסייע בהפחתת התקפות של מניעת שירות המשתמשות במסגרות נתונים ארוכות. אנו יכולים לשנות את אורך המסגרת ליישום על ידי ציון האורך המקסימלי ב application.conf:

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

אנו יכולים גם להגדיר אפשרות תצורה זו על ידי ציון האורך המקסימלי כפרמטר של שורת פקודה:

sbt -Dwebsocket.frame.maxLength = 64k ריצה

7.2. פסק זמן לחיבור סרק

כברירת מחדל, השחקן בו אנו משתמשים לטיפול ב- WebSocket מסתיים לאחר דקה אחת. הסיבה לכך היא ששרת ה- Play שבו היישום שלנו פועל הוא בעל זמן קצוב לתפוגה של 60 דקות ללא סרק. המשמעות היא שכל החיבורים שאינם מקבלים בקשה תוך שישים שניות נסגרים אוטומטית.

אנו יכולים לשנות זאת באמצעות אפשרויות תצורה. בואו ניגש אל שלנו application.conf ולשנות את השרת ללא פסק זמן סרק:

play.server.http.idleTimeout = "אינסופי"

לחלופין נוכל להעביר את האפשרות כארגומנטים של שורת פקודה:

sbt -Dhttp.idleTimeout = אינסוף ריצה

אנו יכולים גם להגדיר זאת על ידי ציון devSettings ב build.sbt.

אפשרויות תצורה שצוינו ב build.sbt משמשים רק בפיתוח, הם יתעלמו מההפקה:

PlayKeys.devSettings + = "play.server.http.idleTimeout" -> "אינסופי"

אם נפעיל מחדש את היישום, השחקן לא יסתיים.

אנו יכולים לשנות את הערך לשניות:

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

אנו יכולים לברר מידע נוסף על אפשרויות התצורה הזמינות בתיעוד Play Framework.

8. מסקנה

במדריך זה יישמנו את WebSockets במסגרת Play Play עם שחקני Akka וזרמי Akka.

לאחר מכן המשכנו לבדוק כיצד להשתמש ישירות בשחקני Akka ואז ראינו כיצד ניתן להגדיר זרמי Akka לטיפול בחיבור WebSocket.

בצד הלקוח, השתמשנו ב- JavaScript כדי לטפל באירועי WebSocket שלנו.

לבסוף, בדקנו כמה אפשרויות תצורה בהן אנו יכולים להשתמש.

כרגיל, קוד המקור של הדרכה זו זמין ב- GitHub.