// ĠCopyright Ġ 2 0 2 0 ĠThe ĠCross - Media ĠMeasurement ĠAuthors Ċ // Ċ // Ġ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 Ġorg . wf an et . measurement . du ch y . service . system . v 1 alpha Ċ Ċ import Ġcom . google . protobuf . ByteString Ċ import Ġjava . util . logging . Logger Ċ import Ġkotlinx . coroutines . flow . Flow Ċ import Ġkotlinx . coroutines . flow . map Ċ import Ġorg . wf an et . measurement . common . Cons umed Flow Item Ċ import Ġorg . wf an et . measurement . common . consume First Ċ import Ġorg . wf an et . measurement . common . grpc . fail Grpc Ċ import Ġorg . wf an et . measurement . common . grpc . grpc Require Ċ import Ġorg . wf an et . measurement . common . grpc . grpc Require NotNull Ċ import Ġorg . wf an et . measurement . common . identity . D uch y Identity Ċ import Ġorg . wf an et . measurement . common . identity . du ch y Identity From Context Ċ import Ġorg . wf an et . measurement . du ch y . storage . Computation Blob Context Ċ import Ġorg . wf an et . measurement . du ch y . storage . Computation Store Ċ import Ġorg . wf an et . measurement . internal . du ch y . Advance Computation Request Ġas ĠAsync Advance Computation Request Ċ import Ġorg . wf an et . measurement . internal . du ch y . Async Computation Control Grpc Kt . Async Computation Control Coroutine Stub Ċ import Ġorg . wf an et . measurement . storage . Storage Client Ċ import Ġorg . wf an et . measurement . system . v 1 alpha . Advance Computation Request Ċ import Ġorg . wf an et . measurement . system . v 1 alpha . Advance Computation Response Ċ import Ġorg . wf an et . measurement . system . v 1 alpha . Computation Control Grpc Kt . Computation Control Coroutine Impl Base Ċ import Ġorg . wf an et . measurement . system . v 1 alpha . Computation Key Ċ Ċ class ĠComp utation Control Service ( ĊĠ Ġprivate Ġval Ġasync Computation Control Client : ĠAsync Computation Control Coroutine Stub , ĊĠ Ġprivate Ġval Ġcomputation Store : ĠComp utation Store , ĊĠ Ġprivate Ġval Ġd uch y Identity Provider : Ġ() Ġ-> ĠD uch y Identity Ġ= Ġ:: du ch y Identity From Context Ċ ) Ġ: ĠComp utation Control Coroutine Impl Base () Ġ{ ĊĊĠ Ġconstructor ( ĊĠĠĠ Ġasync Computation Control Client : ĠAsync Computation Control Coroutine Stub , ĊĠĠĠ Ġstorage Client : ĠStorage Client , ĊĠĠĠ Ġd uch y Identity Provider : Ġ() Ġ-> ĠD uch y Identity Ġ= Ġ:: du ch y Identity From Context ĊĠ Ġ) Ġ: Ġthis ( async Computation Control Client , ĠComp utation Store ( storage Client ), Ġd uch y Identity Provider ) ĊĊĠ Ġoverride Ġsuspend Ġfun Ġadvance Computation ( ĊĠĠĠ Ġrequests : ĠFlow < Advance Computation Request > ĊĠ Ġ): ĠAdv ance Computation Response Ġ{ ĊĠĠĠ Ġgrpc Require NotNull ( requests . consume First ()) Ġ{ Ġ" Empty Ġrequest Ġstream " Ġ}. use Ġ{ ĊĠĠĠĠĠ Ġconsumed : ĠCon su med Flow Item < Advance Computation Request > Ġ-> ĊĠĠĠĠĠ Ġval Ġheader Ġ= Ġconsumed . item . header ĊĠĠĠĠĠ Ġval Ġglobal Computation Id Ġ= ĊĠĠĠĠĠĠĠ Ġgrpc Require NotNull ( Computation Key . from Name ( header . name )?. computation Id ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġ" Resource Ġname Ġunspecified Ġor Ġinvalid ." ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġgrpc Require ( consumed . has Remaining ) Ġ{ Ġ" Request Ġstream Ġhas Ġno Ġbody " Ġ} ĊĠĠĠĠĠ Ġif Ġ( header . is For Async Computation ()) Ġ{ ĊĠĠĠĠĠĠĠ Ġhandle Async Request ( ĊĠĠĠĠĠĠĠĠĠ Ġheader , ĊĠĠĠĠĠĠĠĠĠ Ġconsumed . remaining . map Ġ{ Ġit . body Chunk . partial Data Ġ}, ĊĠĠĠĠĠĠĠĠĠ Ġglobal Computation Id ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠ Ġfail Grpc Ġ{ Ġ" S ynchronous Ġcomput ations Ġare Ġnot Ġyet Ġsupported ." Ġ} ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠĠĠ Ġreturn ĠAdv ance Computation Response . getDefaultInstance () ĊĠ Ġ} ĊĊĠ Ġ/** ĠWrite Ġthe Ġpayload Ġdata Ġstream Ġas Ġa Ġblob , Ġand Ġadvance Ġthe Ġstage Ġvia Ġthe Ġasync Ġservice . Ġ*/ ĊĠ Ġprivate Ġsuspend Ġfun Ġhandle Async Request ( ĊĠĠĠ Ġheader : ĠAdv ance Computation Request . Header , ĊĠĠĠ Ġcontent : ĠFlow < ByteString >, ĊĠĠĠ Ġglobal Id : ĠString ĊĠ Ġ) Ġ{ ĊĠĠĠ Ġ// ĠTODO : Ġmaybe Ġearly Ġreject Ġthe Ġrequest , Ġe . g ., Ġby Ġchecking Ġif Ġa Ġcomputation Ġwith Ġthis Ġid ĊĠĠĠ Ġ// Ġ Ġexists . ĊĠĠĠ Ġval Ġstage Ġ= Ġheader . stage Expecting Input () ĊĠĠĠ Ġval Ġblob Ġ= Ġcomputation Store . write ( Computation Blob Context ( global Id , Ġstage ), Ġcontent ) ĊĠĠĠ Ġval Ġadvance Async Computation Request Ġ= ĊĠĠĠĠĠ ĠAsync Advance Computation Request . newBuilder () ĊĠĠĠĠĠĠĠ Ġ. apply Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġglobal Computation Id Ġ= Ġglobal Id ĊĠĠĠĠĠĠĠĠĠ Ġcomputation Stage Ġ= Ġstage ĊĠĠĠĠĠĠĠĠĠ Ġblob Path Ġ= Ġblob . blob Key ĊĠĠĠĠĠĠĠĠĠ Ġdata Origin Ġ= Ġd uch y Identity Provider (). id ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠ Ġasync Computation Control Client . advance Computation ( advance Async Computation Request ) ĊĠ Ġ} ĊĊĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠ Ġprivate Ġval Ġlogger : ĠLogger Ġ= ĠLogger . getLogger ( this :: class . java . name ) ĊĠ Ġ} Ċ } Ċ