/* Ċ Ġ* ĠCopyright Ġ 2 0 1 9 Ġ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 Ġcom . squareup . wire . internal Ċ Ċ import Ġcom . squareup . wire . Grpc Response Ċ import Ġcom . squareup . wire . Message Source Ċ import Ġcom . squareup . wire . Proto Adapter Ċ import Ġcom . squareup . wire . use Ċ import Ġokhttp 3 . Call Ċ import Ġokhttp 3 . Callback Ċ import Ġok io . IOException Ċ import Ġjava . util . concurrent . Linked Blocking Deque Ċ Ċ /** Ċ Ġ* ĠThis Ġmessage Ġsource Ġuses Ġa Ġ[ Linked Blocking Deque ] Ġto Ġconnect Ġa Ġreading Ġsource Ġwith Ġa Ġwriting Ċ Ġ* Ġresponse Ġcallback . Ċ Ġ* Ċ Ġ* ĠIt Ġuses Ġtwo Ġsentinel Ġtypes : Ċ Ġ* Ċ Ġ* Ġ Ġ* ĠFailure : Ġen queued Ġwhen Ġthere 's Ġan Ġexception Ġon Ġthe Ġstream . Ċ Ġ* Ġ Ġ* ĠComplete : Ġen queued Ġwhen Ġthe Ġstream Ġcompletes Ġnormally . Ċ Ġ*/ Ċ internal Ġclass ĠBlock ing Message Source < R Ġ: ĠAny >( ĊĠ Ġval Ġresponse Adapter : ĠProto Adapter < R >, ĊĠ Ġval Ġcall : ĠCall Ċ ) Ġ: ĠMessage Source < R > Ġ{ ĊĠ Ġprivate Ġval Ġqueue Ġ= ĠLinked Blocking Deque < Any >( 1 ) ĊĊĠ Ġoverride Ġfun Ġread (): ĠR ? Ġ{ ĊĠĠĠ Ġreturn Ġwhen Ġ( val Ġresult Ġ= Ġqueue . take ()) Ġ{ ĊĠĠĠĠĠ Ġis ĠComplete Ġ-> Ġ{ ĊĠĠĠĠĠĠĠ Ġqueue . put ( result ) Ġ// ĠReplace Ġit . ĊĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġis ĠFailure Ġ-> Ġ{ ĊĠĠĠĠĠĠĠ Ġqueue . put ( result ) Ġ// ĠReplace Ġit . ĊĠĠĠĠĠĠĠ Ġthrow Ġresult . e ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġelse Ġ-> Ġresult Ġas ĠR ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġclose () Ġ{ ĊĠĠĠ Ġcall . cancel () Ġ// ĠShort - circuit Ġthe Ġrequest Ġstream Ġif Ġit 's Ġstill Ġactive . ĊĠ Ġ} ĊĊĠ Ġ/** ĠRead Ġmessages Ġfrom Ġthe Ġresponse Ġbody Ġand Ġwrite Ġthem Ġto Ġthe Ġdeque . Ġ*/ ĊĠ Ġfun Ġread From ResponseBody Callback (): ĠCallback Ġ{ ĊĠĠĠ Ġreturn Ġobject Ġ: ĠCallback Ġ{ ĊĠĠĠĠĠ Ġoverride Ġfun Ġon Failure ( call : ĠCall , Ġe : ĠIOException ) Ġ{ ĊĠĠĠĠĠĠĠ Ġqueue . put ( Failure ( e )) ĊĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠ Ġoverride Ġfun Ġon Response ( call : ĠCall , Ġresponse : ĠGr pc Response ) Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġresponse . use Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresponse . message Source ( response Adapter ). use Ġ{ Ġreader Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmessage Ġ= Ġreader . read () Ġ?: Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġqueue . put ( message ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġexception Ġ= Ġresponse . grpc Response To Exception () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( exception Ġ!= Ġnull ) Ġthrow Ġexception ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠ Ġqueue . put ( Complete ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠIOException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġcall . cancel () Ġ// ĠBreak Ġthe Ġrequest Ġstream Ġif Ġthe Ġresponse Ġstream Ġbreaks . ĊĠĠĠĠĠĠĠĠĠ Ġqueue . put ( Failure ( e )) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġprivate Ġobject ĠComplete ĊĠ Ġprivate Ġclass ĠFailure ( val Ġe : ĠIOException ) Ċ } Ċ