package Ġcom . q ing zh u . message server . service Ċ Ċ import Ġcom . q ing zh u . common . domain . shared . msg . constant . Creator Type Ċ import Ġcom . q ing zh u . common . domain . shared . msg . dto . Message Dto Ċ import Ġcom . q ing zh u . common . domain . shared . msg . dto . Update Message Ċ import Ġcom . q ing zh u . common . domain . shared . msg . value . Message Ċ import Ġcom . q ing zh u . common . util . toJson Ċ import Ġcom . q ing zh u . message server . domain . entity . ChatMessage PO Ċ import Ġcom . q ing zh u . message server . domain . entity . Conversation Status Ċ import Ġcom . q ing zh u . message server . domain . value . Customer To Staff Ċ import Ġcom . q ing zh u . message server . mapper . ChatMessage Mapper Ċ import Ġcom . q ing zh u . message server . repository . ChatMessage PO Repository Ċ import Ġorg . springframework . cloud . stream . function . Stream Bridge Ċ import Ġorg . springframework . data . redis . core . Reactive Redis Template Ċ import Ġorg . springframework . data . redis . core . Reactive Z Set Operations Ċ import Ġorg . springframework . messaging . support . Message Builder Ċ import Ġorg . springframework . stereotype . Service Ċ import Ġreactor . core . publisher . Flux Ċ import Ġreactor . core . publisher . Mono Ċ import Ġreactor . core . scheduler . Schedulers Ċ import Ġreactor . kotlin . core . publisher . to Mono Ċ import Ġjava . time . Duration Ċ import Ġjava . util . concurrent . TimeUnit Ċ Ċ @ Service Ċ class ĠMessage Service ( ĊĠĠĠ Ġprivate Ġval Ġcustomer Status Service : ĠCustomer Status Service , ĊĠĠĠ Ġprivate Ġval Ġstaff Status Service : ĠStaff Status Service , ĊĠĠĠ Ġprivate Ġval Ġredis Template : ĠReactive Redis Template < String , ĠString >, ĊĠĠĠ Ġprivate Ġval Ġstream Bridge : ĠStream Bridge , ĊĠĠĠ Ġprivate Ġval Ġchat Message PO Repository : ĠChat Message PO Repository , ĊĠĠĠ Ġprivate Ġval Ġconversation Status Service : ĠCon versation Status Service , Ċ ) Ġ{ ĊĠĠĠ Ġprivate Ġval Ġz Set : ĠReactive Z Set Operations < String , ĠString > Ġ= Ġredis Template . ops For Z Set () ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġ å®¢ æľį / å®¢æĪ· ç¦» çº¿ ä¸Ģ æĸ¹ ç¦» çº¿ çĬ¶æĢģ ä»į çĦ¶ åı¯ä»¥ åıĳéĢģ æ¶Īæģ¯ ĊĠĠĠĠ Ġ* Ġæ¯ı æ¬¡ å®¢ æľį çĻ»å½ķ æĪĸ å®¢æĪ· æīĵå¼Ģ åĴ ¨ è¯¢ çķĮéĿ¢ æĹ¶ ï¼Į æĭī èµ· åĲĮæŃ¥ åºĵ æľĢæĸ° æ¶Īæģ¯ ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġfun ĠMono < Customer To Staff >. sync Message ( message Dto : ĠMessage Dto ): ĠMono < Boolean > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmessage Ġ= Ġmessage Dto . message ĊĊĠĠĠĠĠĠĠ Ġval Ġfrom Ġ= ĠMono . just OrEmpty ( message . from ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġ" msg :${ message . organization Id }:${ message . creator Type . name . toLowerCase ()} :${ it }" Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġto Ġ= ĠMono . just (" msg :${ message . organization Id }:${ message . type . name . toLowerCase ()} :${ message . to }") ĊĊĠĠĠĠĠĠĠ Ġlateinit Ġvar Ġstaff Update Message : ĠUpdate Message ĊĠĠĠĠĠĠĠ Ġlateinit Ġvar Ġcustomer Update Message : ĠUpdate Message ĊĠĠĠĠĠĠĠ Ġval Ġstaff Ġ= Ġthis . flatMap Ġ{ ĠMono . just OrEmpty ( it . staff Id ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġstaff Id Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstaff Status Service . consistency Update ( message . organization Id !! , Ġstaff Id ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmap Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstaff Update Message Ġ= ĠUpdate Message ( it . pts Ġ?: Ġmessage . seq Id , Ġmessage , Ġmessage Dto . clientId ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( message . type Ġ== ĠCreator Type . STA FF ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . pts Ġ= Ġmessage . seq Id ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġcustomer Ġ= Ġthis . flatMap Ġ{ ĠMono . just OrEmpty ( it . customer Id ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġcustomer Status Service . findBy UserId ( message . organization Id !! , Ġit ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Next Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcustomer Update Message Ġ= ĠUpdate Message ( it . pts Ġ?: Ġmessage . seq Id , Ġmessage , Ġmessage Dto . clientId ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. cache () ĊĊĠĠĠĠĠĠĠ Ġval Ġdata Ġ= Ġmessage . toJson () ĊĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠ Ġ* Ġ åħĪ åĨĻ åĪ° Ġredis Ġ åĨį è¿Ľè¡Į æİ¨ éĢģ ï¼Į çĽ¸ æ¯Ķè¾ĥ äºİ åħĪ æİ¨ éĢģ åĨį åĨĻ redis ï¼Į å»¶ è¿ Ł èĻ½çĦ¶ å¢ŀåĬł äºĨ ĊĠĠĠĠĠĠĠĠ Ġ* Ġ ä½Ĩæĺ¯ çĶ¨æĪ· æ¯ıæ¬¡ çĻ» éĻĨ åĲİ æ£ĢæŁ¥ redis å¤§ æ¦Ĥ çİĩ ä¸įä¼ļ ä¸ ¢ å¤± æķ°æį® ï¼Ī åı¯èĥ½ä¼ļ æľī éĩįå¤į ï¼ī ĊĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠ Ġreturn Ġto ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. concat With ( from ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġkey Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmessage . to Mono () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġ è¿ĩæ»¤ Ġ ç³»ç»Ł æ¶Īæģ¯ ï¼Į è¿Ļä¸ª ç±»åŀĭçļĦ æ¶Īæģ¯ æ²¡ å¿ħè¦ģ åĨĻ åĲĮæŃ¥ åºĵ äºĨ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġit . creator Type Ġ!= ĠCreator Type . SYS Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġ åĨĻ æī© æķ£ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġz Set . add ( key , Ġdata , Ġmessage . seq Id . toDouble ()) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġredis Template . expire ( key , ĠDuration . of Days ( 3 )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. then ( this ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Many Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcustomer . flatMap Iterable Ġ{ Ġcs Ġ-> Ġcs . client Access Server Map . entries Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. concat With ( staff . flatMap Iterable Ġ{ Ġstaff Status Ġ-> Ġstaff Status . client Access Server Map . entries Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġentry Ġ-> Ġentry . key Ġ!= Ġmessage Dto . clientId Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġentry Ġ-> Ġentry . value Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġ æİ¨ éĢģ åĪ° çĻ» éĻĨ çļĦ åħ¶ä»ĸ å®¢æĪ·ç«¯ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. distinct () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Next Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġ åıĳéĢģ æ¶Īæģ¯ åĪ° kafka ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( message . type ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCreator Type . CUSTOMER Ġ-> Ġstream Bridge . send ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ"${ it }. message ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMessage Builder . with Payload ( customer Update Message . toJson ()). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCreator Type . STA FF Ġ-> Ġstream Bridge . send ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ"${ it }. message ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMessage Builder . with Payload ( staff Update Message . toJson ()). build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. collect List () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. publish On ( Schedulers . bounded Elastic ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġ æĮģ ä¹ħ åĮĸ Ġ åĪ° Ġ æķ°æį®åºĵ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcustomer . flatMap Ġ{ Ġcs Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġchat Message Ġ= ĠChat Message Mapper . mapper . map To From Message ( message Dto . message ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsave Message ( ChatMessage PO ( chat Message , Ġcs . userId )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. then ( customer ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġ åıĳéĢģ å®Į æ¶Īæģ¯ è¿Ľè¡Į ç»Łè®¡ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . statistics By ChatMessage ( message ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcustomer Status Service . rewrite Status ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconversation Status Service . statistics By ChatMessage ( it . organization Id , Ġit . userId , Ġmessage ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. thenReturn ( true ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġæ ¶ Ī æģ¯ åıĳéĢģ æŃ¥éª¤ ï¼ļ ĊĠĠĠĠ Ġ* Ġ 1 ãĢģ ä¿ĿåŃĺ åĪ° Ġredis Ġz Set Ġ( åĨĻ æī© æķ£ / åıĳéĢģ æĸ¹ ï¼Į æİ¥åıĹ æĸ¹ éĥ½ä¼ļ åĨĻåħ¥ ) ĊĠĠĠĠ Ġ* Ġ 2 ãĢģ æŁ¥æī¾ æ¶Īæģ¯ æİ¥åıĹ æĸ¹ Ġredis Ġ è®¢ éĺħ åľ°åĿĢ ï¼Į åĲĳ Ġredis Ġ åĪĹ éĺŁ åĨĻåħ¥ æ¶Īæģ¯ Ġ( æİ¨ éĢģ æ¨¡å¼ı ) ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġsend ( message Dto : ĠMono < Message Dto >): ĠMono < Boolean > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġmessage Dto ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. publish On ( Schedulers . bounded Elastic ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( it . message . creator Type ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġ å°ģè£ħ å®¢ æľį å®¢æĪ· Ġid Ġå¯¹ ï¼Į çĦ¶åĲİ è¿ĩæ»¤ åıĳéĢģ æ¶Īæģ¯ çļĦ å®¢æĪ·ç«¯ ID ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCreator Type . CUSTOMER Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( it . message . type ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCreator Type . STA FF Ġ-> ĠCustomer To Staff ( it . message . from , Ġit . message . to ). to Mono () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. sync Message ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> ĠMono . just ( false ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCreator Type . STA FF Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( it . message . type ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCreator Type . CUSTOMER Ġ-> ĠCustomer To Staff ( it . message . to , Ġit . message . from ). to Mono () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. sync Message ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCreator Type . GROUP Ġ-> ĠTODO (" åıĳéĢģ ç»Ļ ç¾¤ ç»Ħ ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCreator Type . SYS Ġ-> ĠTODO (" åıĳéĢģ ç³»ç»Ł æ¶Īæģ¯ ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> ĠMono . just ( false ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCreator Type . SYS Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( it . message . type ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCreator Type . CUSTOMER Ġ-> ĠCustomer To Staff ( it . message . to , Ġnull ). to Mono (). sync Message ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCreator Type . STA FF Ġ-> ĠCustomer To Staff ( null , Ġit . message . to ). to Mono (). sync Message ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> ĠMono . just ( false ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> ĠMono . just ( false ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġbot Ġæ ¶ Ī æģ¯ å¤ĦçĲĨ ï¼Į åĨĻåħ¥ åĪ° çĶ¨æĪ· çļĦ Ġredis Ġz Set Ġ åĪĹè¡¨ ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġsync Bot Message ( message : ĠMessage ): ĠMono < Boolean > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġis Bot Ġ= Ġmessage . creator Type Ġ== ĠCreator Type . STA FF ĊĠĠĠĠĠĠĠ Ġval ĠuserId Ġ= Ġif Ġ( is Bot ) Ġmessage . to Ġelse Ġmessage . from ĊĠĠĠĠĠĠĠ Ġval Ġkey Ġ= Ġ" msg :${ message . organization Id }:${ Creator Type . CUSTOMER . name . toLowerCase ()} :${ userId }" ĊĠĠĠĠĠĠĠ Ġval Ġchat Message Ġ= ĠChat Message Mapper . mapper . map To From Message ( message ) ĊĠĠĠĠĠĠĠ Ġreturn ĠMono ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. zip ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġz Set . add ( key , Ġmessage . toJson (), Ġmessage . seq Id . toDouble ()), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsave Message ( ChatMessage PO ( chat Message , ĠuserId !! )), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcustomer Status Service . findBy UserId ( message . organization Id !! , ĠuserId ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. publish On ( Schedulers . bounded Elastic ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġæł¹æį® æ¶Īæģ¯ ç»Łè®¡ ä¸ĢäºĽ æķ°æį® ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . statistics By ChatMessage ( message ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcustomer Status Service . rewrite Status ( it , Ġ 1 5 , ĠTimeUnit . MIN UTES ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconversation Status Service . statistics By ChatMessage ( it . organization Id , Ġit . userId , Ġmessage ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. thenReturn ( true ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit . t 1 Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġsuspend Ġfun Ġtransform To () Ġ{ ĊĠĠĠĠĠĠĠ ĠTODO (" è½¬ æİ¥ å®¢æĪ· ï¼Į æĬĬ å®¢æĪ· èģ Ĭ å¤© ä¿¡æģ¯ æī¹éĩı åĨĻåħ¥ åĪ° å®¢ æľį çļĦ Ġz Set ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġwebsocket åıĳéĢģ ĠAssignment Ġ äºĭä»¶ ç»Ļ å®¢ æľį ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġsend Assignment Event ( conversation Status Dto : ĠMono < Conversation Status >): ĠMono < Unit > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġconversation Status Dto Cache Ġ= Ġconversation Status Dto . cache (). checkpoint (" æ£ĢæŁ¥ status ") ĊĠĠĠĠĠĠĠ Ġreturn Ġconversation Status Dto Cache ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġit . interaction Ġ== Ġ 1 Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Many Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstaff Status Service . find Staff ( it . organization Id , Ġit . staff Id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Iterable Ġ{ Ġcs Ġ-> Ġcs . client Access Server Map . entries Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġentry Ġ-> Ġentry . value Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. distinct () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. transform Ġ{ ĠFlux . zip ( it , Ġconversation Status Dto Cache ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Next Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġ åıĳéĢģ æ¶Īæģ¯ åĪ° kafka ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstream Bridge . send ("${ it . t 1 }. conv ", ĠMessage Builder . with Payload ( it . t 2 . toJson ()). build ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. collect List () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ ĠMono . empty () Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġsend To All My Customers ( message Dto : ĠMono < Message Dto >) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġstaff Status Ġ= Ġmessage Dto ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmessage Ġ= Ġit . message ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstaff Status Service . find Staff ( message . organization Id !! , Ġmessage . from !! ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġit . is Off Line () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Iterable Ġ{ Ġit . userId List Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmessage Dto . do On Next Ġ{ Ġmsg Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg . message . to Ġ= Ġit ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġsend ( it . to Mono ()) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġ æĮģ ä¹ħ åĮĸ åįķ çĭ¬ æ¶Īæģ¯ åĪ° Ġc assandra ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġfun Ġsave Message ( chat Message PO : ĠChat Message PO ): ĠMono < ChatMessage PO > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġconversation Status Service . find Latest By UserId ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġchat Message PO . chat Message Key . organization Id , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġchat Message PO . chat Message Key . owner Id . to Long () ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġchat Message PO . conversation Id Ġ= Ġit . id ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġchat Message PO Repository . save ( chat Message PO ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ }