/* Ċ Ġ* ĠCopyright Ġ 2 0 1 6 ĠNet flix , Ġ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 Ġio . r socket . android . internal Ċ Ċ import Ġio . reactivex . Comp letable Ċ import Ġio . reactivex . Flow able Ċ import Ġio . reactivex . processors . Behavior Processor Ċ import Ġio . r socket . android . Dup lex Connection Ċ import Ġio . r socket . android . Frame Ċ import Ġio . r socket . android . Frame Type .* Ċ import Ġio . r socket . android . Frame Type . SETUP Ċ import Ġio . r socket . android . plugins . Dup lex Connection Interceptor . Type Ċ import Ġio . r socket . android . plugins . Dup lex Connection Interceptor . Type .* Ċ import Ġio . r socket . android . plugins . Interceptor Registry Ċ import Ġorg . reactiv est ream s . Publisher Ċ import Ġorg . slf 4 j . LoggerFactory Ċ Ċ internal Ġclass ĠServer Connection Dem ux er ( source : ĠD up lex Connection , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġplugins : ĠInter ceptor Registry ) ĊĠĠĠ Ġ: ĠConnection Dem ux er ( source , Ġplugins ) Ġ{ ĊĊĠĠĠ Ġoverride Ġfun Ġdem ux ( frame : ĠFrame ): ĠType Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġstream Id Ġ= Ġframe . stream Id ĊĠĠĠĠĠĠĠ Ġval Ġtype Ġ= Ġframe . type ĊĠĠĠĠĠĠĠ Ġreturn Ġwhen Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ( type Ġ== ĠSET UP ) Ġor Ġ( type Ġ== ĠRES UME ) Ġ-> ĠType . SETUP ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ( stream Id Ġ> Ġ 0 ) Ġ-> Ġif Ġ(( stream Id Ġand Ġ 1 Ġ== Ġ 1 )) ĠRESP ON DER Ġelse ĠREQUEST ER ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ( type Ġ== ĠMET ADATA _ PUSH ) Ġ-> ĠRESP ON DER ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> ĠSERVICE ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ internal Ġclass ĠClient Connection Dem ux er ( source : ĠD up lex Connection , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġplugins : ĠInter ceptor Registry ) ĊĠĠĠ Ġ: ĠConnection Dem ux er ( source , Ġplugins ) Ġ{ ĊĊĠĠĠ Ġoverride Ġfun Ġdem ux ( frame : ĠFrame ): ĠType Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġstream Id Ġ= Ġframe . stream Id ĊĠĠĠĠĠĠĠ Ġval Ġtype Ġ= Ġframe . type ĊĠĠĠĠĠĠĠ Ġreturn Ġwhen Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ( type Ġ== ĠSET UP ) Ġor Ġ( type Ġ== ĠRES UME ) Ġ-> ĠType . SETUP ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstream Id Ġ> Ġ 0 Ġ-> Ġif Ġ(( stream Id Ġand Ġ 1 Ġ== Ġ 1 )) ĠREQUEST ER Ġelse ĠRESP ON DER ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ( type Ġ== ĠMET ADATA _ PUSH ) Ġ-> ĠRESP ON DER ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> ĠSERVICE ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ sealed Ġclass ĠConnection Dem ux er ( private Ġval Ġsource : ĠD up lex Connection , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġplugins : ĠInter ceptor Registry ) Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġsetup Connection : ĠD up lex Connection ĊĠĠĠ Ġprivate Ġval Ġresponder Connection : ĠD up lex Connection ĊĠĠĠ Ġprivate Ġval Ġrequest er Connection : ĠD up lex Connection ĊĠĠĠ Ġprivate Ġval Ġservice Connection : ĠD up lex Connection ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġsrc Ġ= Ġplugins . intercept Connection ( ALL , Ġsource ) ĊĊĠĠĠĠĠĠĠ Ġval Ġsetup Conn Ġ= ĠDem ux ed Connection ( src ) ĊĠĠĠĠĠĠĠ Ġsetup Connection Ġ= Ġplugins . intercept Connection ( Type . SETUP , Ġsetup Conn ) ĊĊĠĠĠĠĠĠĠ Ġval Ġrequest er Conn Ġ= ĠDem ux ed Connection ( src ) ĊĠĠĠĠĠĠĠ Ġrequest er Connection Ġ= Ġplugins . intercept Connection ( REQUEST ER , Ġrequest er Conn ) ĊĊĠĠĠĠĠĠĠ Ġval Ġresponder Conn Ġ= ĠDem ux ed Connection ( src ) ĊĠĠĠĠĠĠĠ Ġresponder Connection Ġ= Ġplugins . intercept Connection ( RESP ON DER , Ġresponder Conn ) ĊĊĠĠĠĠĠĠĠ Ġval Ġservice Conn Ġ= ĠDem ux ed Connection ( src ) ĊĠĠĠĠĠĠĠ Ġservice Connection Ġ= Ġplugins . intercept Connection ( SERVICE , Ġservice Conn ) ĊĊĠĠĠĠĠĠĠ Ġsrc . receive () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. groupBy (:: dem ux ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. subscribe ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ{ Ġinbound Frames Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtype Ġ= Ġinbound Frames . key ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġconnection Ġ= Ġwhen Ġ( type ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠType . SETUP Ġ-> Ġsetup Conn ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠRESP ON DER Ġ-> Ġresponder Conn ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠREQUEST ER Ġ-> Ġrequest er Conn ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠSERVICE Ġ-> Ġservice Conn ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġall NotSupported ( type ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconnection . accept ( inbound Frames ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ{ Ġ/* noop Ġ- Ġerrors Ġare Ġhandled Ġby Ġdem ux ed Ġframe Ġstreams */ Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġabstract Ġfun Ġdem ux ( frame : ĠFrame ): ĠType ĊĊĠĠĠ Ġfun Ġresponder Connection (): ĠD up lex Connection Ġ= Ġresponder Connection ĊĊĠĠĠ Ġfun Ġrequest er Connection (): ĠD up lex Connection Ġ= Ġrequest er Connection ĊĊĠĠĠ Ġfun Ġsetup Connection (): ĠD up lex Connection Ġ= Ġsetup Connection ĊĊĠĠĠ Ġfun Ġservice Connection (): ĠD up lex Connection Ġ= Ġservice Connection ĊĊĠĠĠ Ġfun Ġclose (): ĠComp letable Ġ= Ġsource . close () ĊĊĠĠĠ Ġprivate Ġfun Ġall NotSupported ( type : ĠType ?): ĠDem ux ed Connection Ġ{ ĊĠĠĠĠĠĠĠ Ġthrow ĠIllegalStateException (" dem ux er Ġdoes Ġnot Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" support Ġframes Ġof Ġtype : Ġ$ type ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġclass ĠDem ux ed Connection ( private Ġval Ġsender : ĠD up lex Connection ) ĊĠĠĠĠĠĠĠ Ġ: ĠD up lex Connection Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġdebug Enabled : ĠBoolean Ġ= ĠLOGGER . isDebugEnabled ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġreceiv ers Ġ= ĠBehavior Processor . create < Flow able < Frame >>() ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġreceiver Ġ= Ġreceiv ers . first Or Error () ĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġsend ( frame : ĠPublisher < Frame >): ĠComp letable Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġframes Ġ= Ġif Ġ( debug Enabled ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠFlow able . from Publisher ( frame ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. do On Next Ġ{ Ġf Ġ-> ĠLOGGER . debug (" sending Ġ-> Ġ" Ġ+ Ġf . toString ()) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġframe ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġsender . send ( frames ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġfun Ġaccept ( receiver : ĠFlow able < Frame >) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreceiv ers . onNext ( receiver ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġsend One ( frame : ĠFrame ): ĠComp letable Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( debug Enabled ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLOGGER . debug (" sending Ġ-> Ġ" Ġ+ Ġframe . toString ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġsender . send One ( frame ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġreceive (): ĠFlow able < Frame > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġreceiver . flatMap Publisher Ġ{ Ġf Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( debug Enabled ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġf . do On Next Ġ{ Ġframe Ġ-> ĠLOGGER . debug (" re ceiving Ġ-> Ġ" Ġ+ Ġframe . toString ()) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġf ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġclose (): ĠComp letable Ġ= Ġsender . close () ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun ĠonClose (): ĠComp letable Ġ= Ġsender . onClose () ĊĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġavailability (): ĠDouble Ġ= Ġsender . availability () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval ĠLOGGER Ġ= ĠLoggerFactory . getLogger (" io . r socket . Frame Logger ") ĊĠĠĠ Ġ} Ċ } ĊĊ