package Ġbur p ĊĊ Ċ class ĠStream ( val Ġconnection : ĠConnection , Ġval Ġstream ID : ĠInt , Ġval Ġreq : ĠRequest , Ġfrom Client : ĠBoolean ) Ġ{ ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġconst Ġval ĠC LEAN Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġconst Ġval ĠW ANT BODY Ġ= Ġ 1 ĊĠĠĠĠĠĠĠ Ġconst Ġval ĠDONE Ġ= Ġ 2 ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġvar Ġstate Ġ= ĠStream . CLEAN ĊĠĠĠ Ġvar Ġheaders : ĠString Ġ= Ġ"" ĊĠĠĠ Ġvar Ġbody : ĠString Ġ= Ġ"" ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( from Client ) Ġ{ Ċ // ĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġframe Ġ= ĠFrame ( 0 x 0 4 , Ġ 0 x 0 0 , Ġ 0 , ĠByteArray ( 0 )) Ċ // ĠĠĠĠĠĠĠĠĠĠĠ Ġconnection . send Frame ( frame ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠException (" Can 't Ġhandle Ġclient - initi ated Ġstreams Ġyet ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġfun Ġprocess Frame ( frame Bytes : ĠByteArray ) Ġ{ ĊĠĠĠĠĠĠĠ Ġconnection . state Lock . read Lock (). lock () ĊĠĠĠĠĠĠĠ Ġval Ġframe Ġ= Ġparse Frame ( frame Bytes , Ġstream ID ) ĊĊĠĠĠĠĠĠĠ ĠConnection . debug (" Stream Ġ$ stream ID Ġtype Ġ${ frame . type } Ġflags Ġ${ frame . flags }") ĊĠĠĠĠĠĠĠ Ġif Ġ( frame Ġis ĠSettings Frame ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( frame . max Concurrent Streams Ġ!= Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthis Ġis Ġhard - coded Ġinstead ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠConnection . debug (" Change Ġmax Ġconcurrent Ġstreams Ġfrom Ġ${ connection . requests Per Connection } Ġto Ġ${ frame . max Concurrent Streams }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconnection . max Concurrent Streams Ġ= Ġframe . max Concurrent Streams ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġif Ġit 's Ġnot Ġan Ġack , Ġrespond Ġwith Ġan Ġack ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( frame . payload . size Ġ!= Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġack Frame Ġ= ĠFrame ( 0 x 0 4 , Ġ 0 x 0 1 , Ġ 0 , Ġbyte ArrayOf ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconnection . send Frame ( ack Frame ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġtodo Ġnot Ġsure Ġif Ġthis Ġshould Ġbe Ġtriggered Ġon Ġthe ĠSET T INGS Ġor Ġthe ĠACK ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconnection . start Sending Requests () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġtodo Ġin Ġtheory Ġthis Ġis Ġsafe Ġbecause ... Ġwe Ġwon 't Ġget Ġa Ġreply ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconnection . streams . remove ( stream ID ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// return ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġif Ġ( frame Ġis ĠGo Away Frame ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠUtils . out (" Server Ġsent Ġa ĠGO AW AY , Ġdump ing Ġthe Ġwhole Ġconnection ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconnection . state Lock . read Lock (). unlock () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconnection . close () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġif Ġ( frame Ġis ĠR st Stream Frame ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠUtils . out (" Server Ġsent Ġa ĠR ST _ STREAM , Ġdump ing Ġthe Ġwhole Ġconnection ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconnection . state Lock . read Lock (). unlock () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconnection . close () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġif Ġ( state Ġ== ĠC LEAN ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( frame Ġis ĠHeader Frame ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheaders Ġ= Ġframe . header String ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! frame . die ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstate Ġ= ĠW ANT BODY ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstate Ġ= ĠDONE ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġif Ġ( frame Ġis ĠDataFrame ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠException (" Expected Ġheaders , Ġgot Ġdata ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġif Ġ( state Ġ== ĠW ANT BODY ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( frame Ġis ĠDataFrame ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbody Ġ+= Ġframe . body ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ċ Ċ // ĠĠĠĠĠĠĠ Ġif Ġ( connection . state Ġ== ĠConnection . CLOSED ) Ġ{ Ċ // ĠĠĠĠĠĠĠĠĠĠĠ ĠUtils . out (" Blocked Ġattempt Ġto Ġremove Ġstream Ġfrom Ġclosed Ġconnection ...") Ċ // ĠĠĠĠĠĠĠĠĠĠĠ Ġconnection . state Lock . read Lock (). unlock () Ċ // ĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ċ // ĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġif Ġ( frame . die ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreq . time Ġ= Ġ( System . nano Time () Ġ- Ġreq . time ) Ġ/ Ġ 1 0 0 0 0 0 0 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconnection . engine . successful Requests . get And Increment () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( Th readed Request Engine . should G zip ( headers )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbody Ġ= ĠTh readed Request Engine . de compress ( body . toByteArray ( Charsets . ISO _ 8 8 5 9 _ 1 )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġreq . response Ġ= Ġheaders Ġ+ Ġ"\ r \ n " Ġ+ Ġbody ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġinteresting Ġ= Ġconnection . engine . process Response ( req , Ġ( req . response Ġas ĠString ). toByteArray ( Charsets . ISO _ 8 8 5 9 _ 1 )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconnection . engine . invoke Callback ( req , Ġinteresting ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠConnection . debug (" Deleting Ġstream Ġ$ stream ID ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconnection . streams . remove ( stream ID ) ĊĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġconnection . state Lock . read Lock (). unlock () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġparse Frame ( raw : ĠByteArray , Ġstream ID : ĠInt ): ĠFrame Ġ{ ĊĠĠĠĠĠĠĠ Ġ// Connection . debug (" Raw : Ġ"+ raw . asList ()) ĊĊĠĠĠĠĠĠĠ Ġval Ġtype Ġ= Ġraw [ 3 ] ĊĠĠĠĠĠĠĠ Ġval Ġflags Ġ= Ġraw [ 4 ] ĊĠĠĠĠĠĠĠ Ġval Ġpayload Ġ= Ġraw . slice Array ( 9 .. raw . size - 1 ) ĊĊĠĠĠĠĠĠĠ Ġif Ġ( stream ID Ġ!= ĠHTTP 2 Utils . four Byte Int ( raw . slice Array ( 5 .. 8 ))) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠException (" bad Ġstream ĠID ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ// val Ġstream ID Ġ= Ġfour Byte Int ( raw . slice Array ( 5 .. 8 )) ĊĠĠĠĠĠĠĠ Ġ// Connection . debug (" Type : Ġ"+ type ) ĊĠĠĠĠĠĠĠ Ġ// Connection . debug (" Flags / Stream : Ġ$ flags /$ stream ID ") ĊĠĠĠĠĠĠĠ Ġreturn Ġwhen ( type . toInt ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠ 0 Ġ-> ĠDataFrame ( type , Ġflags , Ġstream ID , Ġpayload ) ĊĠĠĠĠĠĠĠĠĠĠĠĠ 1 Ġ-> ĠHeader Frame ( type , Ġflags , Ġstream ID , Ġpayload , Ġthis ) ĊĠĠĠĠĠĠĠĠĠĠĠĠ 3 Ġ-> ĠR st Stream Frame ( type , Ġflags , Ġstream ID , Ġpayload ) ĊĠĠĠĠĠĠĠĠĠĠĠĠ 4 Ġ-> ĠSettings Frame ( type , Ġflags , Ġstream ID , Ġpayload ) ĊĠĠĠĠĠĠĠĠĠĠĠĠ 6 Ġ-> ĠPing Frame ( type , Ġflags , Ġstream ID , Ġpayload ) ĊĠĠĠĠĠĠĠĠĠĠĠĠ 7 Ġ-> ĠGo Away Frame ( type , Ġflags , Ġstream ID , Ġpayload ) ĊĠĠĠĠĠĠĠĠĠĠĠĠ 8 Ġ-> ĠWindow Frame ( type , Ġflags , Ġstream ID , Ġpayload ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> ĠFrame ( type , Ġflags , Ġstream ID , Ġpayload ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĊ Ċ } Ċ