package Ġch . max ant . k dc . mf . web . boundary Ċ Ċ import Ġch . max ant . k dc . mf . library . Context Ċ import Ġio . small ry e . mut iny . Multi Ċ import Ġio . small ry e . mut iny . subscription . Multi Emitter Ċ import Ġorg . eclipse . micro profile . openapi . annotations . tags . Tag Ċ import Ġorg . jboss . logging . Logger Ċ import Ġorg . jboss . rest easy . annotations . S se ElementType Ċ import Ġjava . util . concurrent . Copy On Write ArrayList Ċ import Ġjavax . enterprise . context . Application Scoped Ċ import Ġjavax . inject . Inject Ċ import Ġjavax . servlet . http . HttpServletResponse Ċ import Ġjavax . ws . rs .* Ċ import Ġjavax . ws . rs . core . MediaType Ċ import Ġjavax . ws . rs . core . Response ĊĊ Ċ @ Application Scoped Ċ @ Path ("/ web ") Ċ @ Tag ( name Ġ= Ġ" web ") Ċ @ Cons umes ( MediaType . APPLICATION _ JSON ) Ċ @ Produces ( MediaType . APPLICATION _ JSON ) Ċ class ĠWeb Resource Ġ{ ĊĊĠĠĠ Ġval Ġlog : ĠLogger Ġ= ĠLogger . getLogger ( this . java Class ) ĊĊĠĠĠ Ġ@ Inject ĊĠĠĠ Ġlateinit Ġvar Ġcontext : ĠContext ĊĊĠĠĠ Ġ// ĠTODO Ġtidy Ġthe Ġentries Ġup Ġwhen Ġthey Ġare Ġno Ġlonger Ġin Ġuse ! Ġtip : Ġsee Ġis Cancelled Ġbelow Ġ- Ġal tho Ġthey re Ġalready Ġremoved Ġwith Ġont ermin ate Ġat Ġthe Ġbottom ? ĊĠĠĠ Ġval Ġsubscriptions Ġ= ĠCopy On Write ArrayList < Emitter State >() ĊĊĠĠĠ Ġfun Ġsend To Sub scribers ( sessionId : ĠString , Ġjson : ĠString , Ġkey : ĠString ) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġsubscriptions ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġit . is Expired Or Cancelled () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized ( it . emitter ) Ġ{ Ġ// ĠTODO Ġis Ġthis Ġnecessary ? Ġdoes Ġit Ġh urt ?? ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" closing Ġcancelled / expired Ġemitter Ġfor Ġsession Ġ${ it . sessionId }. Ġcancelled : Ġ" ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ+ Ġ"${ it . emitter . is Cancelled }, Ġexpired : Ġ${ it . is Expired () }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . emitter . complete () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriptions . remove ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġsubscriptions ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġit . sessionId Ġ== ĠsessionId Ġ|| Ġit . matches ( key ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġ! it . is Expired Or Cancelled () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsynchronized ( it ) Ġ{ Ġ// ĠTODO Ġis Ġthis Ġnecessary ? Ġdoes Ġit Ġh urt ?? ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" emit ting Ġsession Ġ${ it . sessionId } Ġto Ġsubscriber : Ġ$ json . Ġcontext . sessionId Ġis Ġ${ context . getSession Id S af ely (). sessionId }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . touch () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . emitter . emit ( json ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ GET ĊĠĠĠ Ġ@ Path ("/ stream /{ sessionId }") ĊĠĠĠ Ġ@ Produces ( MediaType . SERVER _ SENT _ EVENTS ) ĊĠĠĠ Ġ@ S se ElementType ( MediaType . APPLICATION _ JSON ) ĊĠĠĠ Ġfun Ġstream (@ Path Param (" sessionId ") ĠsessionId : ĠString , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ@ QueryParam (" regex ") Ġregex String : ĠString ?, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ@ javax . ws . rs . core . Context Ġresponse : ĠHttpServletResponse ): ĠMulti < String > Ġ{ ĊĠĠĠĠĠĠĠ Ġ// Ġhttps :// server fault . com / questions / 8 0 1 6 2 8 / for - server - sent - events - sse - what - nginx - proxy - configuration - is - appro priate ĊĠĠĠĠĠĠĠ Ġresponse . setHeader (" Cache - Control ", Ġ" no - cache ") ĊĠĠĠĠĠĠĠ Ġresponse . setHeader (" X - Accel - Buffer ing ", Ġ" no ") ĊĠĠĠĠĠĠĠ Ġreturn ĠMulti . createFrom () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. emitter Ġ{ Ġe : ĠMulti Emitter < in ĠString ?> Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriptions . add ( Emitter State ( e , ĠsessionId , Ġregex String )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġe . on Termination Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġe . complete () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" re moving Ġterm ind ated Ġsubscription Ġ$ sessionId ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriptions . remove If Ġ{ Ġit . sessionId Ġ== ĠsessionId Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġ// ĠTODO Ġif Ġwe Ġget Ġmemory Ġproblems , Ġadd Ġa Ġdifferent ĠBack Pressure Strategy Ġas Ġa Ġsecond Ġparameter Ġto Ġthe Ġemitter Ġmethod ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ GET ĊĠĠĠ Ġ@ Path ("/ stats ") ĊĠĠĠ Ġfun Ġstats (): ĠResponse Ġ= ĠResponse . ok (""" ĊĠĠĠĠĠĠĠ Ġ{ Ġ" subscriptions Count ": Ġ${ this . subscriptions . size }, ĊĠĠĠĠĠĠĠĠĠ Ġ" subscriptions ": Ġ${ this . subscriptions . map Ġ{ Ġit . sessionId Ġ}} ĠĊĠĠĠĠĠĠĠ Ġ} """. trim Indent (). replace (" Ġ", Ġ""). replace ("\ r ", Ġ""). replace ("\ n ", Ġ" ")). build () Ċ } Ċ Ċ class ĠEmit ter State ( val Ġemitter : ĠMulti Emitter < in ĠString ? >, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval ĠsessionId : ĠString , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġregex String : ĠString ? Ġ= Ġnull ) Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġregex : ĠRegex ? Ġ= Ġif ( regex String Ġ== Ġnull ) Ġnull Ġelse ĠRegex ( regex String ) ĊĊĠĠĠ Ġprivate Ġvar Ġlast Used : ĠLong Ġ= ĠSystem . currentTimeMillis () ĊĊĠĠĠ Ġfun Ġis Expired () Ġ= ĠSystem . currentTimeMillis () Ġ- Ġthis . last Used Ġ> ĠF IVE _ MIN UTES ĊĊĠĠĠ Ġfun Ġis Expired Or Cancelled () Ġ= Ġthis . emitter . is Cancelled Ġ|| Ġis Expired () ĊĊĠĠĠ Ġfun Ġtouch () Ġ{ ĊĠĠĠĠĠĠĠ Ġlast Used Ġ= ĠSystem . currentTimeMillis () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġmatches ( toMatch : ĠString ): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġif ( regex Ġ== Ġnull ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġto Match . matches ( regex ) ĊĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠF IVE _ MIN UTES Ġ= Ġ 5 Ġ* Ġ 6 0 Ġ* Ġ 1 _ 0 0 0 ĊĠĠĠ Ġ} Ċ }