/* Ċ Ġ* ĠCopyright ĠAmazon . com , ĠInc . Ġor Ġits Ġaffiliates . ĠAll ĠRights ĠReserved . Ċ Ġ* ĠSPDX - License - Identifier : ĠApache - 2 . 0 . Ċ Ġ*/ Ċ Ċ package Ġaws . sdk . kotlin . runtime . http . engine . crt Ċ Ċ import Ġaws . sdk . kotlin . crt . CRT Ċ import Ġaws . sdk . kotlin . crt . http .* Ċ import Ġaws . sdk . kotlin . crt . io . Buffer Ċ import Ġaws . sdk . kotlin . runtime . ClientException Ċ import Ġaws . smithy . kotlin . runtime . http .* Ċ import Ġaws . smithy . kotlin . runtime . http . Headers Builder Ċ import Ġaws . smithy . kotlin . runtime . http . response . HttpResponse Ċ import Ġaws . smithy . kotlin . runtime . io . Sdk Byte Read Channel Ċ import Ġaws . smithy . kotlin . runtime . logging . Logger Ċ import Ġkotlinx . atomic fu . locks . re entrant Lock Ċ import Ġkotlinx . atomic fu . locks . with Lock Ċ import Ġkotlinx . coroutines . Experimental Cor outines Api Ċ import Ġkotlinx . coroutines . channels . Channel Ċ Ċ /** Ċ Ġ* ĠImplements Ġthe ĠC RT Ġstream Ġresponse Ġinterface Ġwhich Ġproxies Ġthe Ġresponse Ġfrom Ġthe ĠC RT Ġto Ġthe ĠSDK Ċ Ġ* Ġ@ param Ġconn ĠThe ĠHTTP Ġconnection Ġused Ġto Ġmake Ġthe Ġrequest . ĠWill Ġbe Ġclosed Ġwhen Ġthe Ġresponse Ġhandler Ġcompletes Ċ Ġ*/ Ċ @ Opt In ( Experimental Cor outines Api :: class ) Ċ internal Ġclass ĠS dk Stream Response Handler ( ĊĠĠĠ Ġprivate Ġval Ġconn : ĠHttpClient Connection Ċ ) Ġ: ĠHttp Stream Response Handler Ġ{ ĊĠĠĠ Ġ// ĠTODO Ġ- Ġneed Ġto Ġcancel Ġthe Ġstream Ġwhen Ġthe Ġbody Ġis Ġclosed Ġfrom Ġthe Ġcaller Ġside Ġearly . ĊĠĠĠ Ġ// ĠThere Ġis Ġno Ġgreat Ġway Ġto Ġdo Ġthat Ġcurrently Ġwithout Ġeither Ġ( 1 ) Ġclosing Ġthe Ġconnection Ġor Ġ( 2 ) Ġthrowing Ġan ĊĠĠĠ Ġ// Ġexception Ġfrom Ġa Ġcallback Ġsuch Ġthat ĠAWS _ OP _ ERROR Ġis Ġreturned . ĠWait Ġfor ĠHttp Stream Ġto Ġhave Ġexplicit Ġcancellation ĊĊĠĠĠ Ġprivate Ġval Ġlogger Ġ= ĠLogger . getLogger < Sdk Stream Response Handler >() ĊĠĠĠ Ġprivate Ġval Ġresponse Ready Ġ= ĠChannel < HttpResponse >( 1 ) ĊĠĠĠ Ġprivate Ġval Ġheaders Ġ= ĠHeaders Builder () ĊĊĠĠĠ Ġprivate Ġvar Ġsdk Body : ĠBuffered Read Channel ? Ġ= Ġnull ĊĊĠĠĠ Ġprivate Ġval Ġlock Ġ= Ġre entrant Lock () Ġ// Ġprotect s Ġc rt Stream Ġand Ġcancelled Ġstate ĊĠĠĠ Ġprivate Ġvar Ġc rt Stream : ĠHttp Stream ? Ġ= Ġnull ĊĠĠĠ Ġ// Ġif Ġthe Ġ( coroutine ) Ġjob Ġis Ġcompleted Ġbefore Ġthe Ġstream 's Ġon Response Complete Ġcallback Ġis ĊĠĠĠ Ġ// Ġinvoked Ġ( for Ġany Ġreason ) Ġwe Ġconsider Ġthe Ġstream Ġ" cancelled " ĊĠĠĠ Ġprivate Ġvar Ġcancelled Ġ= Ġfalse ĊĊĠĠĠ Ġprivate Ġval ĠInt . is Main Headers Block : ĠBoolean ĊĠĠĠĠĠĠĠ Ġget () Ġ= Ġwhen Ġ( this ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠHttp Header Block . MAIN . block Type Ġ-> Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġfalse ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġvar Ġstream Completed Ġ= Ġfalse ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠCalled Ġby Ġthe Ġresponse Ġread Ġchannel Ġas Ġdata Ġis Ġconsumed ĊĠĠĠĠ Ġ* Ġ@ param Ġsize Ġthe Ġnumber Ġof Ġbytes Ġconsumed ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġfun Ġon Data Cons umed ( size : ĠInt ) Ġ{ ĊĠĠĠĠĠĠĠ Ġlock . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġc rt Stream ?. increment Window ( size ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġon Response Headers ( ĊĠĠĠĠĠĠĠ Ġstream : ĠHttp Stream , ĊĠĠĠĠĠĠĠ Ġresponse StatusCode : ĠInt , ĊĠĠĠĠĠĠĠ Ġblock Type : ĠInt , ĊĠĠĠĠĠĠĠ Ġnext Headers : ĠList < Http Header >? ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ(! block Type . is Main Headers Block ) Ġreturn ĊĊĠĠĠĠĠĠĠ Ġnext Headers ?. forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġheaders . append ( it . name , Ġit . value ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcreate Http ResponseBody ( content Length : ĠLong ): ĠHttp Body Ġ{ ĊĠĠĠĠĠĠĠ Ġsdk Body Ġ= Ġbuffered Read Channel (:: on Data Cons umed ) ĊĠĠĠĠĠĠĠ Ġreturn Ġobject Ġ: ĠHttp Body . Streaming () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġval Ġcontent Length : ĠLong Ġ= Ġcontent Length ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġread From (): ĠS dk Byte Read Channel Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġsdk Body !! ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ// Ġsignal Ġresponse Ġready Ġand Ġengine Ġcan Ġproceed Ġ( all Ġthat Ġis Ġrequired Ġis Ġheaders , Ġbody Ġis Ġconsumed Ġasynchronously ) ĊĠĠĠ Ġprivate Ġfun Ġsignal Response ( stream : ĠHttp Stream ) Ġ{ ĊĠĠĠĠĠĠĠ Ġ// Ġalready Ġsign alled ĊĠĠĠĠĠĠĠ Ġif Ġ( response Ready . is Closed For Send ) Ġreturn ĊĊĠĠĠĠĠĠĠ Ġval Ġtransfer Encoding Ġ= Ġheaders [" Transfer - Encoding "] ?. lowercase () ĊĠĠĠĠĠĠĠ Ġval Ġchunk ed Ġ= Ġtransfer Encoding Ġ== Ġ" chunk ed " ĊĠĠĠĠĠĠĠ Ġval Ġcontent Length Ġ= Ġheaders [" Content - Length "] ?. to Long () Ġ?: Ġ 0 ĊĠĠĠĠĠĠĠ Ġval Ġstatus Ġ= ĠHttpStatusCode . from Value ( stream . response StatusCode ) ĊĊĠĠĠĠĠĠĠ Ġval Ġhas Body Ġ= Ġ( content Length Ġ> Ġ 0 Ġ|| Ġchunk ed ) Ġ&& ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ( status Ġ! in ĠlistOf ( HttpStatusCode . Not Modified , ĠHttpStatusCode . No Content )) Ġ&& ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ! status . is Information al () ĊĊĠĠĠĠĠĠĠ Ġval Ġbody Ġ= Ġwhen Ġ( has Body ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfalse Ġ-> ĠHttp Body . Empty ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtrue Ġ-> Ġcreate Http ResponseBody ( content Length ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġresp Ġ= ĠHttpResponse ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstatus , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġheaders . build (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġbody ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġresponse Ready . try Send ( resp ) ĊĠĠĠĠĠĠĠ Ġcheck ( result . is Success ) Ġ{ Ġ" sign all ing Ġresponse Ġfailed , Ġresult Ġwas : Ġ${ result . exception OrNull () }" Ġ} ĊĠĠĠĠĠĠĠ Ġresponse Ready . close () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġon Response Headers Done ( stream : ĠHttp Stream , Ġblock Type : ĠInt ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ(! block Type . is Main Headers Block ) Ġreturn ĊĠĠĠĠĠĠĠ Ġsignal Response ( stream ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġon ResponseBody ( stream : ĠHttp Stream , Ġbody Bytes In : ĠBuffer ): ĠInt Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġis Cancelled Ġ= Ġlock . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġc rt Stream Ġ= Ġstream ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcancelled ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// Ġshort Ġcircuit , Ġstop Ġbuffer ing Ġdata Ġand Ġdiscard Ġremaining Ġincoming Ġbytes ĊĠĠĠĠĠĠĠ Ġif Ġ( is Cancelled ) Ġreturn Ġbody Bytes In . len ĊĊĠĠĠĠĠĠĠ Ġ// Ġwe Ġshould Ġhave Ġcreated Ġa Ġresponse Ġchannel Ġif Ġwe Ġexpected Ġa Ġbody ĊĠĠĠĠĠĠĠ Ġval Ġsdk Resp Chan Ġ= ĠcheckNotNull ( sdk Body ) Ġ{ Ġ" unexpected Ġresponse Ġbody " Ġ} ĊĠĠĠĠĠĠĠ Ġsdk Resp Chan . write ( body Bytes In ) ĊĊĠĠĠĠĠĠĠ Ġ// Ġexplicit Ġwindow Ġmanagement Ġis Ġdone Ġin ĠBuffered Read Channel Ġwhich Ġcalls Ġ` on Data Cons umed ` ĊĠĠĠĠĠĠĠ Ġ// Ġas Ġdata Ġis Ġread Ġfrom Ġthe Ġchannel ĊĠĠĠĠĠĠĠ Ġreturn Ġ 0 ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġon Response Complete ( stream : ĠHttp Stream , ĠerrorCode : ĠInt ) Ġ{ ĊĠĠĠĠĠĠĠ Ġ// Ġstream Ġis Ġonly Ġvalid Ġuntil Ġthe Ġend Ġof Ġthis Ġcallback , Ġensure Ġany Ġfurther Ġdata Ġbeing Ġread Ġdownstream ĊĠĠĠĠĠĠĠ Ġ// Ġdoesn 't Ġcall Ġincrement Window Ġon Ġa Ġresource Ġthat Ġhas Ġbeen Ġfree 'd ĊĠĠĠĠĠĠĠ Ġlock . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġc rt Stream Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstream Completed Ġ= Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ// Ġclose Ġthe Ġbody Ġchannel ĊĠĠĠĠĠĠĠ Ġif Ġ( errorCode Ġ!= Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġerror Description Ġ= ĠC RT . error String ( errorCode ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġex Ġ= ĠClient Exception (" C rt Http Engine :: response Ġfailed : Ġec =$ errorCode ; Ġdescription =$ error Description ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresponse Ready . close ( ex ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsdk Body ?. cancel ( ex ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġclosing Ġthe Ġchannel Ġto Ġindicate Ġno Ġmore Ġdata Ġwill Ġbe Ġsent ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsdk Body ?. close () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġensure Ġa Ġresponse Ġwas Ġsign alled Ġ( will Ġclose Ġthe Ġchannel Ġon Ġit 's Ġown Ġif Ġit Ġwasn 't Ġalready Ġsent ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsignal Response ( stream ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġinternal Ġsuspend Ġfun ĠwaitFor Response (): ĠHttpResponse Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġresponse Ready . receive () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠInv oked Ġonly Ġafter Ġthe Ġconsumer Ġis Ġfinished Ġwith Ġthe Ġresponse Ġand Ġit Ġis Ġsafe Ġto Ġcleanup Ġresources ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġinternal Ġfun Ġcomplete () Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠWe Ġhave Ġno Ġway Ġof Ġcancell ing Ġthe Ġstream , Ġwe Ġhave Ġto Ġdrive Ġit Ġto Ġex haust ion ĠOR Ġclose Ġthe Ġconnection . ĊĠĠĠĠĠĠĠ Ġ// ĠAt Ġthis Ġpoint Ġwe Ġknow Ġit 's Ġsafe Ġto Ġrelease Ġresources Ġso Ġif Ġthe Ġstream Ġhasn 't Ġcompleted Ġyet ĊĠĠĠĠĠĠĠ Ġ// Ġwe Ġforce fully Ġshutdown Ġthe Ġconnection . ĠThis Ġcan Ġhappen Ġwhen Ġthe Ġstream 's Ġwindow Ġis Ġfull Ġand Ġit 's Ġwaiting ĊĠĠĠĠĠĠĠ Ġ// Ġon Ġthe Ġwindow Ġto Ġbe Ġincrement ed Ġto Ġproceed Ġ( i . e . Ġthe Ġuser Ġdidn 't Ġconsume Ġthe Ġstream Ġfor Ġwhatever Ġreason ĊĠĠĠĠĠĠĠ Ġ// Ġand Ġmore Ġdata Ġis Ġpending Ġar rival ). ĠIt Ġcan Ġalso Ġhappen Ġif Ġthe Ġcoroutine Ġfor Ġthis Ġrequest Ġis Ġcancelled ĊĠĠĠĠĠĠĠ Ġ// Ġbefore Ġon Response Complete Ġfires . ĊĠĠĠĠĠĠĠ Ġlock . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġforce Close Ġ= Ġ! stream Completed ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( force Close ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug Ġ{ Ġ" stream Ġdid Ġnot Ġcomplete Ġbefore Ġjob , Ġforcing Ġconnection Ġshutdown ! Ġhandler =$ this ; Ġconn =$ conn ; Ġstream =$ crt Stream " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconn . shutdown () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcancelled Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġconn . close () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ