/* Ċ Ġ* ĠCopyright Ġ( C ) Ġ 2 0 1 6 ĠSquare , ĠInc . Ċ Ġ* Ċ Ġ* ĠLicensed Ġunder Ġthe ĠApache ĠLicense , ĠVersion Ġ 2 . 0 Ġ( the Ġ" License "); Ċ Ġ* Ġyou Ġmay Ġnot Ġuse Ġthis Ġfile Ġexcept Ġin Ġcompliance Ġwith Ġthe ĠLicense . Ċ Ġ* ĠYou Ġmay Ġobtain Ġa Ġcopy Ġof Ġthe ĠLicense Ġat Ċ Ġ* Ċ Ġ* ĠĠĠĠĠ Ġhttp :// www . apache . org / licenses / LICENSE - 2 . 0 Ċ Ġ* Ċ Ġ* ĠUnless Ġrequired Ġby Ġapplicable Ġlaw Ġor Ġagreed Ġto Ġin Ġwriting , Ġsoftware Ċ Ġ* Ġdistributed Ġunder Ġthe ĠLicense Ġis Ġdistributed Ġon Ġan Ġ" AS ĠIS " ĠBASIS , Ċ Ġ* ĠWITHOUT ĠWARRANTIES ĠOR ĠCONDITIONS ĠOF ĠANY ĠKIND , Ġeither Ġexpress Ġor Ġimplied . Ċ Ġ* ĠSee Ġthe ĠLicense Ġfor Ġthe Ġspecific Ġlanguage Ġgoverning Ġpermissions Ġand Ċ Ġ* Ġlimitations Ġunder Ġthe ĠLicense . Ċ Ġ*/ Ċ package Ġokhttp 3 . internal . ws Ċ Ċ import Ġokhttp 3 . Call Ċ import Ġokhttp 3 . Callback Ċ import Ġokhttp 3 . EventListener Ċ import Ġokhttp 3 . Ok HttpClient Ċ import Ġokhttp 3 . Protocol Ċ import Ġokhttp 3 . Real Call Ċ import Ġokhttp 3 . Request Ċ import Ġokhttp 3 . Response Ċ import Ġokhttp 3 . WebSocket Ċ import Ġokhttp 3 . WebSocket Listener Ċ import Ġokhttp 3 . internal . close Quiet ly Ċ import Ġokhttp 3 . internal . connection . Exchange Ċ import Ġokhttp 3 . internal . thread Factory Ċ import Ġokhttp 3 . internal . ws . WebSocket Protocol . CLOSE _ CLIENT _ GO ING _ AW AY Ċ import Ġokhttp 3 . internal . ws . WebSocket Protocol . CLOSE _ MESSAGE _ MAX Ċ import Ġokhttp 3 . internal . ws . WebSocket Protocol . OPCODE _ BINARY Ċ import Ġokhttp 3 . internal . ws . WebSocket Protocol . OPCODE _ TEXT Ċ import Ġokhttp 3 . internal . ws . WebSocket Protocol . validate Close Code Ċ import Ġok io . Buffered Sink Ċ import Ġok io . Buffered Source Ċ import Ġok io . ByteString Ċ import Ġok io . ByteString . Companion . encode Utf 8 Ċ import Ġok io . ByteString . Companion . to ByteString Ċ import Ġok io . buffer Ċ import Ġjava . io . Closeable Ċ import Ġjava . io . IOException Ċ import Ġjava . net . Protocol Exception Ċ import Ġjava . net . Socket TimeoutException Ċ import Ġjava . util . Array Deque Ċ import Ġjava . util . Random Ċ import Ġjava . util . concurrent . Scheduled ExecutorService Ċ import Ġjava . util . concurrent . Scheduled Future Ċ import Ġjava . util . concurrent . Scheduled ThreadPoolExecutor Ċ import Ġjava . util . concurrent . TimeUnit Ċ import Ġjava . util . concurrent . TimeUnit . MILLISECONDS Ċ Ċ class ĠReal WebSocket ( ĊĠ Ġ/** ĠThe Ġapplication 's Ġoriginal Ġrequest Ġun ad ul ter ated Ġby Ġweb Ġsocket Ġheaders . Ġ*/ ĊĠ Ġprivate Ġval Ġoriginal Request : ĠRequest , ĊĠ Ġinternal Ġval Ġlistener : ĠWebSocket Listener , ĊĠ Ġprivate Ġval Ġrandom : ĠRandom , ĊĠ Ġprivate Ġval Ġping Interval Millis : ĠLong Ċ ) Ġ: ĠWebSocket , ĠWebSocket Reader . Frame Callback Ġ{ ĊĠ Ġprivate Ġval Ġkey : ĠString ĊĊĠ Ġ/** ĠNon - null Ġfor Ġclient Ġweb Ġsockets . ĠThese Ġcan Ġbe Ġcanceled . Ġ*/ ĊĠ Ġprivate Ġvar Ġcall : ĠCall ? Ġ= Ġnull ĊĊĠ Ġ/** ĠThis Ġrunnable Ġprocesses Ġthe Ġoutgoing Ġqueues . ĠCall Ġ[ run Writer ] Ġto Ġafter Ġenqueue ing . Ġ*/ ĊĠ Ġprivate Ġval Ġwriter Runnable : ĠRunnable ĊĊĠ Ġ/** ĠNull Ġuntil Ġthis Ġweb Ġsocket Ġis Ġconnected . ĠOnly Ġaccessed Ġby Ġthe Ġreader Ġthread . Ġ*/ ĊĠ Ġprivate Ġvar Ġreader : ĠWebSocket Reader ? Ġ= Ġnull ĊĊĠ Ġ// ĠAll Ġmutable Ġweb Ġsocket Ġstate Ġis Ġgu arded Ġby Ġthis . ĊĊĠ Ġ/** ĠNull Ġuntil Ġthis Ġweb Ġsocket Ġis Ġconnected . ĠNote Ġthat Ġmessages Ġmay Ġbe Ġen queued Ġbefore Ġthat . Ġ*/ ĊĠ Ġprivate Ġvar Ġwriter : ĠWebSocket Writer ? Ġ= Ġnull ĊĊĠ Ġ/** ĠNull Ġuntil Ġthis Ġweb Ġsocket Ġis Ġconnected . ĠUsed Ġfor Ġwrites , Ġp ings , Ġand Ġclose Ġtimeouts . Ġ*/ ĊĠ Ġprivate Ġvar Ġexecutor : ĠScheduled ExecutorService ? Ġ= Ġnull ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠThe Ġstreams Ġheld Ġby Ġthis Ġweb Ġsocket . ĠThis Ġis Ġnon - null Ġuntil Ġall Ġincoming Ġmessages Ġhave Ġbeen ĊĠĠ Ġ* Ġread Ġand Ġall Ġoutgoing Ġmessages Ġhave Ġbeen Ġwritten . ĠIt Ġis Ġclosed Ġwhen Ġboth Ġreader Ġand Ġwriter Ġare ĊĠĠ Ġ* Ġex hausted , Ġor Ġif Ġthere Ġis Ġany Ġfailure . ĊĠĠ Ġ*/ ĊĠ Ġprivate Ġvar Ġstreams : ĠStreams ? Ġ= Ġnull ĊĊĠ Ġ/** ĠOut going Ġp ongs Ġin Ġthe Ġorder Ġthey Ġshould Ġbe Ġwritten . Ġ*/ ĊĠ Ġprivate Ġval Ġp ong Queue Ġ= ĠArray Deque < ByteString >() ĊĊĠ Ġ/** ĠOut going Ġmessages Ġand Ġclose Ġframes Ġin Ġthe Ġorder Ġthey Ġshould Ġbe Ġwritten . Ġ*/ ĊĠ Ġprivate Ġval Ġmessage And Close Queue Ġ= ĠArray Deque < Any >() ĊĊĠ Ġ/** ĠThe Ġtotal Ġsize Ġin Ġbytes Ġof Ġen queued Ġbut Ġnot Ġyet Ġtransmitted Ġmessages . Ġ*/ ĊĠ Ġprivate Ġvar Ġqueue Size Ġ= Ġ 0 L ĊĊĠ Ġ/** ĠTrue Ġif Ġwe 've Ġen queued Ġa Ġclose Ġframe . ĠNo Ġfurther Ġmessage Ġframes Ġwill Ġbe Ġen queued . Ġ*/ ĊĠ Ġprivate Ġvar Ġen queued Close Ġ= Ġfalse ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠWhen Ġexecuted Ġthis Ġwill Ġcancel Ġthis Ġweb Ġsocket . ĠThis Ġfuture Ġitself Ġshould Ġbe Ġcanceled Ġif Ġthat ĊĠĠ Ġ* Ġis Ġunnecessary Ġbecause Ġthe Ġweb Ġsocket Ġis Ġalready Ġclosed Ġor Ġcanceled . ĊĠĠ Ġ*/ ĊĠ Ġprivate Ġvar Ġcancel Future : ĠScheduled Future < *> ? Ġ= Ġnull ĊĊĠ Ġ/** ĠThe Ġclose Ġcode Ġfrom Ġthe Ġpeer , Ġor Ġ- 1 Ġif Ġthis Ġweb Ġsocket Ġhas Ġnot Ġyet Ġread Ġa Ġclose Ġframe . Ġ*/ ĊĠ Ġprivate Ġvar Ġreceived Close Code Ġ= Ġ- 1 ĊĊĠ Ġ/** ĠThe Ġclose Ġreason Ġfrom Ġthe Ġpeer , Ġor Ġnull Ġif Ġthis Ġweb Ġsocket Ġhas Ġnot Ġyet Ġread Ġa Ġclose Ġframe . Ġ*/ ĊĠ Ġprivate Ġvar Ġreceived Close Reason : ĠString ? Ġ= Ġnull ĊĊĠ Ġ/** ĠTrue Ġif Ġthis Ġweb Ġsocket Ġfailed Ġand Ġthe Ġlistener Ġhas Ġbeen Ġnotified . Ġ*/ ĊĠ Ġprivate Ġvar Ġfailed Ġ= Ġfalse ĊĊĠ Ġ/** ĠTotal Ġnumber Ġof Ġp ings Ġsent Ġby Ġthis Ġweb Ġsocket . Ġ*/ ĊĠ Ġprivate Ġvar Ġsent Ping Count Ġ= Ġ 0 ĊĊĠ Ġ/** ĠTotal Ġnumber Ġof Ġp ings Ġreceived Ġby Ġthis Ġweb Ġsocket . Ġ*/ ĊĠ Ġprivate Ġvar Ġreceived Ping Count Ġ= Ġ 0 ĊĊĠ Ġ/** ĠTotal Ġnumber Ġof Ġp ongs Ġreceived Ġby Ġthis Ġweb Ġsocket . Ġ*/ ĊĠ Ġprivate Ġvar Ġreceived P ong Count Ġ= Ġ 0 ĊĊĠ Ġ/** ĠTrue Ġif Ġwe Ġhave Ġsent Ġa Ġping Ġthat Ġis Ġstill Ġawait ing Ġa Ġreply . Ġ*/ ĊĠ Ġprivate Ġvar Ġawait ing P ong Ġ= Ġfalse ĊĊĠ Ġinit Ġ{ ĊĠĠĠ Ġrequire (" GET " Ġ== Ġoriginal Request . method ) Ġ{ ĊĠĠĠĠĠ Ġ" Request Ġmust Ġbe ĠGET : Ġ${ original Request . method }" ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġthis . key Ġ= ĠByteArray ( 1 6 ). apply Ġ{ Ġrandom . next Bytes ( this ) Ġ}. to ByteString (). base 6 4 () ĊĠĠĠ Ġthis . writer Runnable Ġ= ĠRunnable Ġ{ ĊĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠ Ġwhile Ġ( write One Frame ()) Ġ{ ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠIOException ) Ġ{ ĊĠĠĠĠĠĠĠ Ġfail WebSocket ( e , Ġnull ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġrequest (): ĠRequest Ġ= Ġoriginal Request ĊĊĠ Ġ@ Synchron ized Ġoverride Ġfun Ġqueue Size (): ĠLong Ġ= Ġqueue Size ĊĊĠ Ġoverride Ġfun Ġcancel () Ġ{ ĊĠĠĠ Ġcall !!. cancel () ĊĠ Ġ} ĊĊĠ Ġfun Ġconnect ( client : ĠOk HttpClient ) Ġ{ ĊĠĠĠ Ġval Ġweb Socket Client Ġ= Ġclient . newBuilder () ĊĠĠĠĠĠĠĠ Ġ. event Listener ( EventListener . NONE ) ĊĠĠĠĠĠĠĠ Ġ. protocols ( ONLY _ HTTP 1 ) ĊĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠ Ġval Ġrequest Ġ= Ġoriginal Request . newBuilder () ĊĠĠĠĠĠĠĠ Ġ. header (" Upgrade ", Ġ" websocket ") ĊĠĠĠĠĠĠĠ Ġ. header (" Connection ", Ġ" Upgrade ") ĊĠĠĠĠĠĠĠ Ġ. header (" Sec - WebSocket - Key ", Ġkey ) ĊĠĠĠĠĠĠĠ Ġ. header (" Sec - WebSocket - Version ", Ġ" 1 3 ") ĊĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠ Ġcall Ġ= ĠReal Call . new Real Call ( web Socket Client , Ġrequest , Ġfor WebSocket Ġ= Ġtrue ) ĊĠĠĠ Ġcall !!. enqueue ( object Ġ: ĠCallback Ġ{ ĊĠĠĠĠĠ Ġoverride Ġfun Ġon Response ( call : ĠCall , Ġresponse : ĠResponse ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġexchange Ġ= Ġresponse . exchange ĊĠĠĠĠĠĠĠ Ġval Ġstreams : ĠStreams ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġcheck Upgrade Success ( response , Ġexchange ) ĊĠĠĠĠĠĠĠĠĠ Ġstreams Ġ= Ġexchange !!. new WebSocket Streams () ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠIOException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġexchange ?. web Socket Upgrade Failed () ĊĠĠĠĠĠĠĠĠĠ Ġfail WebSocket ( e , Ġresponse ) ĊĠĠĠĠĠĠĠĠĠ Ġresponse . close Quiet ly () ĊĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// ĠProcess Ġall Ġweb Ġsocket Ġmessages . ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġval Ġname Ġ= Ġ" Ok Http ĠWebSocket Ġ${ request . url . red act () }" ĊĠĠĠĠĠĠĠĠĠ Ġinit Reader And Writer ( name , Ġstreams ) ĊĠĠĠĠĠĠĠĠĠ Ġlistener . on Open ( this @ Real WebSocket , Ġresponse ) ĊĠĠĠĠĠĠĠĠĠ Ġloop Reader () ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġfail WebSocket ( e , Ġnull ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠ Ġoverride Ġfun Ġon Failure ( call : ĠCall , Ġe : ĠIOException ) Ġ{ ĊĠĠĠĠĠĠĠ Ġfail WebSocket ( e , Ġnull ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ}) ĊĠ Ġ} ĊĊĠ Ġ@ Throws ( IOException :: class ) ĊĠ Ġinternal Ġfun Ġcheck Upgrade Success ( response : ĠResponse , Ġexchange : ĠExchange ?) Ġ{ ĊĠĠĠ Ġif Ġ( response . code () Ġ!= Ġ 1 0 1 ) Ġ{ ĊĠĠĠĠĠ Ġthrow ĠProtocol Exception ( ĊĠĠĠĠĠĠĠĠĠ Ġ" Expected ĠHTTP Ġ 1 0 1 Ġresponse Ġbut Ġwas Ġ'${ response . code ()} Ġ${ response . message () }' ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġval Ġheader Connection Ġ= Ġresponse . header (" Connection ") ĊĠĠĠ Ġif Ġ(! " Upgrade ". equals ( header Connection , Ġignore Case Ġ= Ġtrue )) Ġ{ ĊĠĠĠĠĠ Ġthrow ĠProtocol Exception ( ĊĠĠĠĠĠĠĠĠĠ Ġ" Expected Ġ' Connection ' Ġheader Ġvalue Ġ' Upgrade ' Ġbut Ġwas Ġ'$ header Connection '") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġval Ġheader Upgrade Ġ= Ġresponse . header (" Upgrade ") ĊĠĠĠ Ġif Ġ(! " websocket ". equals ( header Upgrade , Ġignore Case Ġ= Ġtrue )) Ġ{ ĊĠĠĠĠĠ Ġthrow ĠProtocol Exception ( ĊĠĠĠĠĠĠĠĠĠ Ġ" Expected Ġ' Upgrade ' Ġheader Ġvalue Ġ' websocket ' Ġbut Ġwas Ġ'$ header Upgrade '") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġval Ġheader Accept Ġ= Ġresponse . header (" Sec - WebSocket - Accept ") ĊĠĠĠ Ġval Ġaccept Expected Ġ= Ġ( key Ġ+ ĠWebSocket Protocol . ACCEPT _ MAGIC ). encode Utf 8 (). sha 1 (). base 6 4 () ĊĠĠĠ Ġif Ġ( accept Expected Ġ!= Ġheader Accept ) Ġ{ ĊĠĠĠĠĠ Ġthrow ĠProtocol Exception ( ĊĠĠĠĠĠĠĠĠĠ Ġ" Expected Ġ' Sec - WebSocket - Accept ' Ġheader Ġvalue Ġ'$ accept Expected ' Ġbut Ġwas Ġ'$ header Accept '") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġif Ġ( exchange Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠ Ġthrow ĠProtocol Exception (" Web ĠSocket Ġexchange Ġmissing : Ġbad Ġinterceptor ?") ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ@ Throws ( IOException :: class ) ĊĠ Ġfun Ġinit Reader And Writer ( name : ĠString , Ġstreams : ĠStreams ) Ġ{ ĊĠĠĠ Ġsynchronized ( this ) Ġ{ ĊĠĠĠĠĠ Ġthis . streams Ġ= Ġstreams ĊĠĠĠĠĠ Ġthis . writer Ġ= ĠWebSocket Writer ( streams . client , Ġstreams . sink , Ġrandom ) ĊĠĠĠĠĠ Ġthis . executor Ġ= ĠScheduled ThreadPoolExecutor ( 1 , Ġthread Factory ( name , Ġfalse )) ĊĠĠĠĠĠ Ġif Ġ( ping Interval Millis Ġ!= Ġ 0 L ) Ġ{ ĊĠĠĠĠĠĠĠ Ġexecutor !!. schedule At Fixed Rate ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠPing Runnable (), Ġping Interval Millis , Ġping Interval Millis , ĠMIL LI SECONDS ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġif Ġ(! message And Close Queue . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠ Ġrun Writer () Ġ// ĠSend Ġmessages Ġthat Ġwere Ġen queued Ġbefore Ġwe Ġwere Ġconnected . ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġreader Ġ= ĠWebSocket Reader ( streams . client , Ġstreams . source , Ġthis ) ĊĠ Ġ} ĊĊĠ Ġ/** ĠReceive Ġframes Ġuntil Ġthere Ġare Ġno Ġmore . ĠInv oked Ġonly Ġby Ġthe Ġreader Ġthread . Ġ*/ ĊĠ Ġ@ Throws ( IOException :: class ) ĊĠ Ġfun Ġloop Reader () Ġ{ ĊĠĠĠ Ġwhile Ġ( received Close Code Ġ== Ġ- 1 ) Ġ{ ĊĠĠĠĠĠ Ġ// ĠThis Ġmethod Ġcall Ġresults Ġin Ġone Ġor Ġmore Ġon Read * Ġmethods Ġbeing Ġcalled Ġon Ġthis Ġthread . ĊĠĠĠĠĠ Ġreader !!. process Next Frame () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠFor Ġtesting : Ġreceive Ġa Ġsingle Ġframe Ġand Ġreturn Ġtrue Ġif Ġthere Ġare Ġmore Ġframes Ġto Ġread . ĠInv oked ĊĠĠ Ġ* Ġonly Ġby Ġthe Ġreader Ġthread . ĊĠĠ Ġ*/ ĊĠ Ġ@ Throws ( IOException :: class ) ĊĠ Ġfun Ġprocess Next Frame (): ĠBoolean Ġ{ ĊĠĠĠ Ġreturn Ġtry Ġ{ ĊĠĠĠĠĠ Ġreader !!. process Next Frame () ĊĠĠĠĠĠ Ġreceived Close Code Ġ== Ġ- 1 ĊĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠ Ġfail WebSocket ( e , Ġnull ) ĊĠĠĠĠĠ Ġfalse ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĠFor Ġtesting : Ġwait Ġuntil Ġthe Ġweb Ġsocket 's Ġexecutor Ġhas Ġterminated . Ġ*/ ĊĠ Ġ@ Throws ( InterruptedException :: class ) ĊĠ Ġfun Ġawait Termination ( timeout : ĠInt , Ġtime Unit : ĠTimeUnit ) Ġ{ ĊĠĠĠ Ġexecutor !!. await Termination ( timeout . to Long (), Ġtime Unit ) ĊĠ Ġ} ĊĊĠ Ġ/** ĠFor Ġtesting : Ġforce Ġthis Ġweb Ġsocket Ġto Ġrelease Ġits Ġthreads . Ġ*/ ĊĠ Ġ@ Throws ( InterruptedException :: class ) ĊĠ Ġfun ĠtearDown () Ġ{ ĊĠĠĠ Ġif Ġ( cancel Future Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠ Ġcancel Future !!. cancel ( false ) ĊĠĠĠ Ġ} ĊĠĠĠ Ġexecutor !!. shutdown () ĊĠĠĠ Ġexecutor !!. await Termination ( 1 0 , ĠTimeUnit . SECONDS ) ĊĠ Ġ} ĊĊĠ Ġ@ Synchron ized Ġfun Ġsent Ping Count (): ĠInt Ġ= Ġsent Ping Count ĊĊĠ Ġ@ Synchron ized Ġfun Ġreceived Ping Count (): ĠInt Ġ= Ġreceived Ping Count ĊĊĠ Ġ@ Synchron ized Ġfun Ġreceived P ong Count (): ĠInt Ġ= Ġreceived P ong Count ĊĊĠ Ġ@ Throws ( IOException :: class ) ĊĠ Ġoverride Ġfun Ġon Read Message ( text : ĠString ) Ġ{ ĊĠĠĠ Ġlistener . on Message ( this , Ġtext ) ĊĠ Ġ} ĊĊĠ Ġ@ Throws ( IOException :: class ) ĊĠ Ġoverride Ġfun Ġon Read Message ( bytes : ĠByte String ) Ġ{ ĊĠĠĠ Ġlistener . on Message ( this , Ġbytes ) ĊĠ Ġ} ĊĊĠ Ġ@ Synchron ized Ġoverride Ġfun Ġon Read Ping ( payload : ĠByte String ) Ġ{ ĊĠĠĠ Ġ// ĠDon 't Ġrespond Ġto Ġp ings Ġafter Ġwe 've Ġfailed Ġor Ġsent Ġthe Ġclose Ġframe . ĊĠĠĠ Ġif Ġ( failed Ġ|| Ġen queued Close Ġ&& Ġmessage And Close Queue . isEmpty ()) Ġreturn ĊĊĠĠĠ Ġp ong Queue . add ( payload ) ĊĠĠĠ Ġrun Writer () ĊĠĠĠ Ġreceived Ping Count ++ ĊĠ Ġ} ĊĊĠ Ġ@ Synchron ized Ġoverride Ġfun Ġon Read P ong ( payload : ĠByte String ) Ġ{ ĊĠĠĠ Ġ// ĠThis ĠAPI Ġdoesn 't Ġexpose Ġp ings . ĊĠĠĠ Ġreceived P ong Count ++ ĊĠĠĠ Ġawait ing P ong Ġ= Ġfalse ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġon Read Close ( code : ĠInt , Ġreason : ĠString ) Ġ{ ĊĠĠĠ Ġrequire ( code Ġ!= Ġ- 1 ) ĊĊĠĠĠ Ġvar Ġto Close : ĠStreams ? Ġ= Ġnull ĊĠĠĠ Ġsynchronized ( this ) Ġ{ ĊĠĠĠĠĠ Ġcheck ( received Close Code Ġ== Ġ- 1 ) Ġ{ Ġ" already Ġclosed " Ġ} ĊĠĠĠĠĠ Ġreceived Close Code Ġ= Ġcode ĊĠĠĠĠĠ Ġreceived Close Reason Ġ= Ġreason ĊĠĠĠĠĠ Ġif Ġ( en queued Close Ġ&& Ġmessage And Close Queue . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠ Ġto Close Ġ= Ġthis . streams ĊĠĠĠĠĠĠĠ Ġthis . streams Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġif Ġ( cancel Future Ġ!= Ġnull ) Ġcancel Future !!. cancel ( false ) ĊĠĠĠĠĠĠĠ Ġthis . executor !!. shutdown () ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠ Ġlistener . on Closing ( this , Ġcode , Ġreason ) ĊĊĠĠĠĠĠ Ġif Ġ( to Close Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠ Ġlistener . on Closed ( this , Ġcode , Ġreason ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠ Ġto Close ?. close Quiet ly () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ// ĠWriter Ġmethods Ġto Ġenqueue Ġframes . ĠThey 'll Ġbe Ġsent Ġasynchronously Ġby Ġthe Ġwriter Ġthread . ĊĊĠ Ġoverride Ġfun Ġsend ( text : ĠString ): ĠBoolean Ġ{ ĊĠĠĠ Ġreturn Ġsend ( text . encode Utf 8 (), ĠOP CODE _ TEXT ) ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġsend ( bytes : ĠByte String ): ĠBoolean Ġ{ ĊĠĠĠ Ġreturn Ġsend ( bytes , ĠOP CODE _ BINARY ) ĊĠ Ġ} ĊĊĠ Ġ@ Synchron ized Ġprivate Ġfun Ġsend ( data : ĠByte String , Ġformat Opcode : ĠInt ): ĠBoolean Ġ{ ĊĠĠĠ Ġ// ĠDon 't Ġsend Ġnew Ġframes Ġafter Ġwe 've Ġfailed Ġor Ġen queued Ġa Ġclose Ġframe . ĊĠĠĠ Ġif Ġ( failed Ġ|| Ġen queued Close ) Ġreturn Ġfalse ĊĊĠĠĠ Ġ// ĠIf Ġthis Ġframe Ġoverflow s Ġthe Ġbuffer , Ġreject Ġit Ġand Ġclose Ġthe Ġweb Ġsocket . ĊĠĠĠ Ġif Ġ( queue Size Ġ+ Ġdata . size Ġ> ĠMAX _ QUEUE _ SIZE ) Ġ{ ĊĠĠĠĠĠ Ġclose ( CLOSE _ CLIENT _ GO ING _ AW AY , Ġnull ) ĊĠĠĠĠĠ Ġreturn Ġfalse ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ// ĠEn queue Ġthe Ġmessage Ġframe . ĊĠĠĠ Ġqueue Size Ġ+= Ġdata . size . to Long () ĊĠĠĠ Ġmessage And Close Queue . add ( Message ( format Opcode , Ġdata )) ĊĠĠĠ Ġrun Writer () ĊĠĠĠ Ġreturn Ġtrue ĊĠ Ġ} ĊĊĠ Ġ@ Synchron ized Ġfun Ġp ong ( payload : ĠByte String ): ĠBoolean Ġ{ ĊĠĠĠ Ġ// ĠDon 't Ġsend Ġp ongs Ġafter Ġwe 've Ġfailed Ġor Ġsent Ġthe Ġclose Ġframe . ĊĠĠĠ Ġif Ġ( failed Ġ|| Ġen queued Close Ġ&& Ġmessage And Close Queue . isEmpty ()) Ġreturn Ġfalse ĊĊĠĠĠ Ġp ong Queue . add ( payload ) ĊĠĠĠ Ġrun Writer () ĊĠĠĠ Ġreturn Ġtrue ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġclose ( code : ĠInt , Ġreason : ĠString ?): ĠBoolean Ġ{ ĊĠĠĠ Ġreturn Ġclose ( code , Ġreason , ĠCANCEL _ AFTER _ CLOSE _ M ILL IS ) ĊĠ Ġ} ĊĊĠ Ġ@ Synchron ized Ġfun Ġclose ( ĊĠĠĠ Ġcode : ĠInt , ĊĠĠĠ Ġreason : ĠString ?, ĊĠĠĠ Ġcancel After Close Millis : ĠLong ĊĠ Ġ): ĠBoolean Ġ{ ĊĠĠĠ Ġvalidate Close Code ( code ) ĊĊĠĠĠ Ġvar Ġreason Bytes : ĠByte String ? Ġ= Ġnull ĊĠĠĠ Ġif Ġ( reason Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠ Ġreason Bytes Ġ= Ġreason . encode Utf 8 () ĊĠĠĠĠĠ Ġrequire ( reason Bytes . size Ġ<= ĠCLOSE _ MESSAGE _ MAX ) Ġ{ ĊĠĠĠĠĠĠĠ Ġ" reason . size () Ġ> Ġ$ CLOSE _ MESSAGE _ MAX : Ġ$ reason " ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġif Ġ( failed Ġ|| Ġen queued Close ) Ġreturn Ġfalse ĊĊĠĠĠ Ġ// ĠIm mediate ly Ġprevent Ġfurther Ġframes Ġfrom Ġbeing Ġen queued . ĊĠĠĠ Ġen queued Close Ġ= Ġtrue ĊĊĠĠĠ Ġ// ĠEn queue Ġthe Ġclose Ġframe . ĊĠĠĠ Ġmessage And Close Queue . add ( Close ( code , Ġreason Bytes , Ġcancel After Close Millis )) ĊĠĠĠ Ġrun Writer () ĊĠĠĠ Ġreturn Ġtrue ĊĠ Ġ} ĊĊĠ Ġprivate Ġfun Ġrun Writer () Ġ{ ĊĠĠĠ Ġassert ( Thread . holds Lock ( this )) ĊĠĠĠ Ġexecutor ?. execute ( writer Runnable ) ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠAt tempts Ġto Ġremove Ġa Ġsingle Ġframe Ġfrom Ġa Ġqueue Ġand Ġsend Ġit . ĠThis Ġpref ers Ġto Ġwrite Ġur gent Ġp ongs ĊĠĠ Ġ* Ġbefore Ġless Ġur gent Ġmessages Ġand Ġclose Ġframes . ĠFor Ġexample Ġit 's Ġpossible Ġthat Ġa Ġcaller Ġwill ĊĠĠ Ġ* Ġenqueue Ġmessages Ġfollowed Ġby Ġp ongs , Ġbut Ġthis Ġsends Ġp ongs Ġfollowed Ġby Ġmessages . ĠP ongs Ġare Ġalways ĊĠĠ Ġ* Ġwritten Ġin Ġthe Ġorder Ġthey Ġwere Ġen queued . ĊĠĠ Ġ* ĊĠĠ Ġ* ĠIf Ġa Ġframe Ġcannot Ġbe Ġsent Ġ- Ġbecause Ġthere Ġare Ġnone Ġen queued Ġor Ġbecause Ġthe Ġweb Ġsocket Ġis Ġnot ĊĠĠ Ġ* Ġconnected Ġ- Ġthis Ġdoes Ġnothing Ġand Ġreturns Ġfalse . ĠOtherwise Ġthis Ġreturns Ġtrue Ġand Ġthe Ġcaller ĊĠĠ Ġ* Ġshould Ġimmediately Ġinvoke Ġthis Ġmethod Ġagain Ġuntil Ġit Ġreturns Ġfalse . ĊĠĠ Ġ* ĊĠĠ Ġ* ĠThis Ġmethod Ġmay Ġonly Ġbe Ġinvoked Ġby Ġthe Ġwriter Ġthread . ĠThere Ġmay Ġbe Ġonly Ġthread Ġinvoking Ġthis ĊĠĠ Ġ* Ġmethod Ġat Ġa Ġtime . ĊĠĠ Ġ*/ ĊĠ Ġ@ Throws ( IOException :: class ) ĊĠ Ġinternal Ġfun Ġwrite One Frame (): ĠBoolean Ġ{ ĊĠĠĠ Ġval Ġwriter : ĠWebSocket Writer ? ĊĠĠĠ Ġval Ġp ong : ĠByte String ? ĊĠĠĠ Ġvar Ġmessage Or Close : ĠAny ? Ġ= Ġnull ĊĠĠĠ Ġvar Ġreceived Close Code Ġ= Ġ- 1 ĊĠĠĠ Ġvar Ġreceived Close Reason : ĠString ? Ġ= Ġnull ĊĠĠĠ Ġvar Ġstreams To Close : ĠStreams ? Ġ= Ġnull ĊĊĠĠĠ Ġsynchronized ( this @ Real WebSocket ) Ġ{ ĊĠĠĠĠĠ Ġif Ġ( failed ) Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġfalse Ġ// ĠFailed Ġweb Ġsocket . ĊĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠ Ġwriter Ġ= Ġthis . writer ĊĠĠĠĠĠ Ġp ong Ġ= Ġp ong Queue . poll () ĊĠĠĠĠĠ Ġif Ġ( p ong Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠ Ġmessage Or Close Ġ= Ġmessage And Close Queue . poll () ĊĠĠĠĠĠĠĠ Ġif Ġ( message Or Close Ġis ĠClose ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġreceived Close Code Ġ= Ġthis . received Close Code ĊĠĠĠĠĠĠĠĠĠ Ġreceived Close Reason Ġ= Ġthis . received Close Reason ĊĠĠĠĠĠĠĠĠĠ Ġif Ġ( received Close Code Ġ!= Ġ- 1 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstreams To Close Ġ= Ġthis . streams ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . streams Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . executor !!. shutdown () ĊĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWhen Ġwe Ġrequest Ġa Ġgrace ful Ġclose Ġalso Ġschedule Ġa Ġcancel Ġof Ġthe Ġweb Ġsocket . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcancel Future Ġ= Ġexecutor !!. schedule ( Cancel Runnable (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ( message Or Close Ġas ĠClose ). cancel After Close Millis , ĠMIL LI SECONDS ) ĊĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( message Or Close Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġreturn Ġfalse Ġ// ĠThe Ġqueue Ġis Ġex hausted . ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠ Ġif Ġ( p ong Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠ Ġwriter !!. write P ong ( p ong ) ĊĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( message Or Close Ġis ĠMessage ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġdata Ġ= Ġ( message Or Close Ġas ĠMessage ). data ĊĠĠĠĠĠĠĠ Ġval Ġsink Ġ= Ġwriter !!. new Message Sink ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ( message Or Close Ġas ĠMessage ). format Opcode , Ġdata . size . to Long ()). buffer () ĊĠĠĠĠĠĠĠ Ġsink . write ( data ) ĊĠĠĠĠĠĠĠ Ġsink . close () ĊĠĠĠĠĠĠĠ Ġsynchronized ( this ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġqueue Size Ġ-= Ġdata . size . to Long () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( message Or Close Ġis ĠClose ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġclose Ġ= Ġmessage Or Close Ġas ĠClose ĊĠĠĠĠĠĠĠ Ġwriter !!. write Close ( close . code , Ġclose . reason ) ĊĊĠĠĠĠĠĠĠ Ġ// ĠWe Ġclosed Ġthe Ġwriter : Ġnow Ġboth Ġreader Ġand Ġwriter Ġare Ġclosed . ĊĠĠĠĠĠĠĠ Ġif Ġ( streams To Close Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġlistener . on Closed ( this , Ġreceived Close Code , Ġreceived Close Reason !! ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠ Ġthrow ĠAssertionError () ĊĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠ Ġreturn Ġtrue ĊĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠ Ġstreams To Close ?. close Quiet ly () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġprivate Ġinner Ġclass ĠPing Runnable Ġ: ĠRunnable Ġ{ ĊĠĠĠ Ġoverride Ġfun Ġrun () Ġ{ ĊĠĠĠĠĠ Ġwrite Ping Frame () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġinternal Ġfun Ġwrite Ping Frame () Ġ{ ĊĠĠĠ Ġval Ġwriter : ĠWebSocket Writer ? ĊĠĠĠ Ġval Ġfailed Ping : ĠInt ĊĠĠĠ Ġsynchronized ( this ) Ġ{ ĊĠĠĠĠĠ Ġif Ġ( failed ) Ġreturn ĊĠĠĠĠĠ Ġwriter Ġ= Ġthis . writer ĊĠĠĠĠĠ Ġfailed Ping Ġ= Ġif Ġ( await ing P ong ) Ġsent Ping Count Ġelse Ġ- 1 ĊĠĠĠĠĠ Ġsent Ping Count ++ ĊĠĠĠĠĠ Ġawait ing P ong Ġ= Ġtrue ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġif Ġ( failed Ping Ġ!= Ġ- 1 ) Ġ{ ĊĠĠĠĠĠ Ġfail WebSocket ( Socket TimeoutException (" sent Ġping Ġbut Ġdidn 't Ġreceive Ġp ong Ġwithin Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠ Ġ"${ ping Interval Millis } ms Ġ( after Ġ${ failed Ping Ġ- Ġ 1 } Ġsuccessful Ġping / p ongs )"), Ġnull ) ĊĠĠĠĠĠ Ġreturn ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠ Ġwriter !!. write Ping ( ByteString . EMPTY ) ĊĠĠĠ Ġ} Ġcatch Ġ( e : ĠIOException ) Ġ{ ĊĠĠĠĠĠ Ġfail WebSocket ( e , Ġnull ) ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġfun Ġfail WebSocket ( e : ĠException , Ġresponse : ĠResponse ?) Ġ{ ĊĠĠĠ Ġval Ġstreams To Close : ĠStreams ? ĊĠĠĠ Ġsynchronized ( this ) Ġ{ ĊĠĠĠĠĠ Ġif Ġ( failed ) Ġreturn Ġ// ĠAlready Ġfailed . ĊĠĠĠĠĠ Ġfailed Ġ= Ġtrue ĊĠĠĠĠĠ Ġstreams To Close Ġ= Ġthis . streams ĊĠĠĠĠĠ Ġthis . streams Ġ= Ġnull ĊĠĠĠĠĠ Ġcancel Future ?. cancel ( false ) ĊĠĠĠĠĠ Ġexecutor ?. shutdown () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠ Ġlistener . on Failure ( this , Ġe , Ġresponse ) ĊĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠ Ġstreams To Close ?. close Quiet ly () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġinternal Ġclass ĠMessage ( ĊĠĠĠ Ġval Ġformat Opcode : ĠInt , ĊĠĠĠ Ġval Ġdata : ĠByte String ĊĠ Ġ) ĊĊĠ Ġinternal Ġclass ĠClose ( ĊĠĠĠ Ġval Ġcode : ĠInt , ĊĠĠĠ Ġval Ġreason : ĠByte String ?, ĊĠĠĠ Ġval Ġcancel After Close Millis : ĠLong ĊĠ Ġ) ĊĊĠ Ġabstract Ġclass ĠStreams ( ĊĠĠĠ Ġval Ġclient : ĠBoolean , ĊĠĠĠ Ġval Ġsource : ĠBuffered Source , ĊĠĠĠ Ġval Ġsink : ĠBuffered Sink ĊĠ Ġ) Ġ: ĠClose able ĊĊĠ Ġinternal Ġinner Ġclass ĠCancel Runnable Ġ: ĠRunnable Ġ{ ĊĠĠĠ Ġoverride Ġfun Ġrun () Ġ{ ĊĠĠĠĠĠ Ġcancel () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠ Ġprivate Ġval ĠONLY _ HTTP 1 Ġ= ĠlistOf ( Protocol . HTTP _ 1 _ 1 ) ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠThe Ġmaximum Ġnumber Ġof Ġbytes Ġto Ġenqueue . ĠR ather Ġthan Ġenqueue ing Ġbeyond Ġthis Ġlimit Ġwe Ġtear Ġdown ĊĠĠĠĠ Ġ* Ġthe Ġweb Ġsocket ! ĠIt 's Ġpossible Ġthat Ġwe 're Ġwriting Ġfaster Ġthan Ġthe Ġpeer Ġcan Ġread . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġconst Ġval ĠMAX _ QUEUE _ SIZE Ġ= Ġ 1 6 L Ġ* Ġ 1 0 2 4 Ġ* Ġ 1 0 2 4 Ġ// Ġ 1 6 ĠM iB . ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠThe Ġmaximum Ġamount Ġof Ġtime Ġafter Ġthe Ġclient Ġcalls Ġ[ close ] Ġto Ġwait Ġfor Ġa Ġgrace ful Ġshutdown . ĠIf ĊĠĠĠĠ Ġ* Ġthe Ġserver Ġdoesn 't Ġrespond Ġthe Ġweb Ġsocket Ġwill Ġbe Ġcanceled . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġconst Ġval ĠCANCEL _ AFTER _ CLOSE _ M ILL IS Ġ= Ġ 6 0 L Ġ* Ġ 1 0 0 0 ĊĠ Ġ} Ċ } Ċ