/* Ċ Ġ* Ġ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 Client Ċ import Ġcom . squareup . wire . Grpc Method Ċ import Ġcom . squareup . wire . Grpc Streaming Call Ċ import Ġcom . squareup . wire . Message Sink Ċ import Ġcom . squareup . wire . Message Source Ċ import Ġjava . util . concurrent . TimeUnit Ċ import Ġkotlinx . coroutines . Coroutine Scope Ċ import Ġkotlinx . coroutines . Dispatch ers Ċ import Ġkotlinx . coroutines . Global Scope Ċ import Ġkotlinx . coroutines . channels . Channel Ċ import Ġkotlinx . coroutines . channels . Receive Channel Ċ import Ġkotlinx . coroutines . channels . Send Channel Ċ import Ġkotlinx . coroutines . launch Ċ import Ġok io . Timeout Ċ Ċ internal Ġclass ĠReal Grpc Streaming Call < S Ġ: ĠAny , ĠR Ġ: ĠAny >( ĊĠ Ġprivate Ġval Ġgrpc Client : ĠGr pc Client , ĊĠ Ġoverride Ġval Ġmethod : ĠGr pc Method < S , ĠR > Ċ ) Ġ: ĠGr pc Streaming Call < S , ĠR > Ġ{ ĊĠ Ġprivate Ġval Ġrequest Body Ġ= Ġnew Dup lex RequestBody () ĊĠ Ġ/** ĠNon - null Ġonce Ġthis Ġis Ġexecuted . Ġ*/ ĊĠ Ġprivate Ġvar Ġcall : ĠCall ? Ġ= Ġnull ĊĠ Ġprivate Ġvar Ġcanceled Ġ= Ġfalse ĊĊĠ Ġoverride Ġval Ġtimeout : ĠTimeout Ġ= ĠL ate Init Timeout () ĊĊĠ Ġinit Ġ{ ĊĠĠĠ Ġtimeout . clear Timeout () ĊĠĠĠ Ġtimeout . clear Deadline () ĊĠ Ġ} ĊĊĠ Ġoverride Ġvar Ġrequest Metadata : ĠMap < String , ĠString > Ġ= Ġmap Of () ĊĊĠ Ġoverride Ġvar Ġresponse Metadata : ĠMap < String , ĠString >? Ġ= Ġnull ĊĠĠĠ Ġinternal Ġset ĊĊĠ Ġoverride Ġfun Ġcancel () Ġ{ ĊĠĠĠ Ġcanceled Ġ= Ġtrue ĊĠĠĠ Ġcall ?. cancel () ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġis Canceled (): ĠBoolean Ġ= Ġcanceled ĊĊĠ Ġoverride Ġfun Ġexecute (): ĠPair < Send Channel < S >, ĠReceive Channel < R >> Ġ= Ġexecute In ( Global Scope ) ĊĊĠ Ġoverride Ġfun Ġexecute In ( scope : ĠCor outine Scope ): ĠPair < Send Channel < S >, ĠReceive Channel < R >> Ġ{ ĊĠĠĠ Ġval Ġrequest Channel Ġ= ĠChannel < S >( 1 ) ĊĠĠĠ Ġval Ġresponse Channel Ġ= ĠChannel < R >( 1 ) ĊĠĠĠ Ġval Ġcall Ġ= Ġinit Call () ĊĊĠĠĠ Ġresponse Channel . invoke On Close Ġ{ ĊĠĠĠĠĠ Ġif Ġ( response Channel . is Closed For Receive ) Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠShort - circuit Ġthe Ġrequest Ġstream Ġif Ġit 's Ġstill Ġactive . ĊĠĠĠĠĠĠĠ Ġcall . cancel () ĊĠĠĠĠĠĠĠ Ġrequest Channel . cancel () ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġscope . launch ( Dispatch ers . IO ) Ġ{ ĊĠĠĠĠĠ Ġrequest Channel . writeTo RequestBody ( ĊĠĠĠĠĠĠĠ Ġrequest Body Ġ= Ġrequest Body , ĊĠĠĠĠĠĠĠ Ġmin Message To Compress Ġ= Ġgrpc Client . min Message To Compress , ĊĠĠĠĠĠĠĠ Ġrequest Adapter Ġ= Ġmethod . request Adapter , ĊĠĠĠĠĠĠĠ Ġcall For Cancel Ġ= Ġcall ĊĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĠĠĠ Ġcall . enqueue ( response Channel . read From ResponseBody Callback ( this , Ġmethod . response Adapter )) ĊĊĠĠĠ Ġreturn Ġrequest Channel Ġto Ġresponse Channel ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġexecute Blocking (): ĠPair < Message Sink < S >, ĠMessage Source < R >> Ġ{ ĊĠĠĠ Ġval Ġcall Ġ= Ġinit Call () ĊĠĠĠ Ġval Ġmessage Source Ġ= ĠBlock ing Message Source ( this , Ġmethod . response Adapter , Ġcall ) ĊĠĠĠ Ġval Ġmessage Sink Ġ= Ġrequest Body . message Sink ( ĊĠĠĠĠĠ Ġmin Message To Compress Ġ= Ġgrpc Client . min Message To Compress , ĊĠĠĠĠĠ Ġrequest Adapter Ġ= Ġmethod . request Adapter , ĊĠĠĠĠĠ Ġcall For Cancel Ġ= Ġcall ĊĠĠĠ Ġ) ĊĠĠĠ Ġcall . enqueue ( message Source . read From ResponseBody Callback ()) ĊĊĠĠĠ Ġreturn Ġmessage Sink Ġto Ġmessage Source ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġis Executed (): ĠBoolean Ġ= Ġcall ?. is Executed () Ġ?: Ġfalse ĊĊĠ Ġoverride Ġfun Ġclone (): ĠGr pc Streaming Call < S , ĠR > Ġ{ ĊĠĠĠ Ġval Ġresult Ġ= ĠReal Grpc Streaming Call ( grpc Client , Ġmethod ) ĊĠĠĠ Ġval Ġold Timeout Ġ= Ġthis . timeout ĊĠĠĠ Ġresult . timeout . also Ġ{ Ġnew Timeout Ġ-> ĊĠĠĠĠĠ Ġnew Timeout . timeout ( old Timeout . timeout Nan os (), ĠTimeUnit . NAN O SECONDS ) ĊĠĠĠĠĠ Ġif Ġ( old Timeout . has Deadline ()) Ġnew Timeout . deadline Nano Time ( old Timeout . deadline Nano Time ()) ĊĠĠĠĠĠ Ġelse Ġnew Timeout . clear Deadline () ĊĠĠĠ Ġ} ĊĠĠĠ Ġreturn Ġresult ĊĠ Ġ} ĊĊĠ Ġprivate Ġfun Ġinit Call (): ĠCall Ġ{ ĊĠĠĠ Ġcheck ( this . call Ġ== Ġnull ) Ġ{ Ġ" already Ġexecuted " Ġ} ĊĊĠĠĠ Ġval Ġresult Ġ= Ġgrpc Client . new Call ( method , Ġrequest Metadata , Ġrequest Body ) ĊĠĠĠ Ġthis . call Ġ= Ġresult ĊĠĠĠ Ġif Ġ( c anceled ) Ġresult . cancel () ĊĠĠĠ Ġ( timeout Ġas ĠL ate Init Timeout ). init ( result . timeout ()) ĊĠĠĠ Ġreturn Ġresult ĊĠ Ġ} Ċ } Ċ