/* Ċ Ġ* ĠCopyright Ġ 2 0 1 9 ĠK roto + ĠContributors Ċ Ġ* Ċ Ġ* Ġ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 . github . mar co f errer . k roto plus . coroutines . integration Ċ Ċ import Ġcom . github . mar co f errer . k roto plus . coroutines . Rpc Call Test Ċ import Ġcom . github . mar co f errer . k roto plus . coroutines . client . client Call Client Streaming Ċ import Ġcom . github . mar co f errer . k roto plus . coroutines . utils . assert Ex Equals Ċ import Ġcom . github . mar co f errer . k roto plus . coroutines . utils . assert Fails Ċ import Ġcom . github . mar co f errer . k roto plus . coroutines . utils . assert Fails With Status 2 Ċ import Ġcom . github . mar co f errer . k roto plus . coroutines . utils . invoke Ċ import Ġcom . github . mar co f errer . k roto plus . coroutines . utils . match Throwable Ċ import Ġcom . github . mar co f errer . k roto plus . coroutines . with Coroutine Context Ċ import Ġio . grpc . Call Options Ċ import Ġio . grpc . Channel Ċ import Ġio . grpc . Client Call Ċ import Ġio . grpc . Client Interceptor Ċ import Ġio . grpc . MethodDescriptor Ċ import Ġio . grpc . Server Intercept ors Ċ import Ġio . grpc . Status Ċ import Ġio . grpc . examples . helloworld . G reet er Coroutine Grpc Ċ import Ġio . grpc . examples . helloworld . G reet er Grpc Ċ import Ġio . grpc . examples . helloworld . Hello Reply Ċ import Ġio . grpc . examples . helloworld . Hello Request Ċ import Ġio . mock k . spy k Ċ import Ġio . mock k . verify Ċ import Ġkotlinx . coroutines . Cancellation Exception Ċ import Ġkotlinx . coroutines . Comp letable Deferred Ċ import Ġkotlinx . coroutines . Coroutine Start Ċ import Ġkotlinx . coroutines . Dispatch ers Ċ import Ġkotlinx . coroutines . Job Ċ import Ġkotlinx . coroutines . cancel Ċ import Ġkotlinx . coroutines . channels . Receive Channel Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . flow . consume As Flow Ċ import Ġkotlinx . coroutines . flow . map Ċ import Ġkotlinx . coroutines . flow . toList Ċ import Ġkotlinx . coroutines . launch Ċ import Ġkotlinx . coroutines . run Blocking Ċ import Ġorg . junit . Test Ċ import Ġjava . util . concurrent . atomic . Atomic Integer Ċ import Ġkotlin . coroutines . coroutine Context Ċ import Ġkotlin . test . assertEquals Ċ import Ġkotlin . test . assertFalse ĊĊ Ċ class ĠClient Streaming Back Pressure Tests Ġ: ĊĠĠĠ ĠRpc Call Test < Hello Request , ĠHello Reply >( G reet er Coroutine Grpc . say Hello Client Streaming Method ) Ġ{ ĊĊĠĠĠ Ġprivate Ġfun Ġsetup Up Server Handler ( ĊĠĠĠĠĠĠĠ Ġblock : Ġsuspend Ġ( request Channel : ĠReceive Channel < Hello Request >) Ġ-> ĠHello Reply ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġservice Impl Ġ= Ġobject Ġ: ĠG reet er Coroutine Grpc . G reet er Impl Base () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġsuspend Ġfun Ġsay Hello Client Streaming ( request Channel : ĠReceive Channel < Hello Request >): ĠHello Reply Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġblock ( request Channel ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġservice Ġ= ĠServer Intercept ors . intercept ( service Impl , Ġcall State ) ĊĠĠĠĠĠĠĠ Ġgrpc Server Rule . service Registry . add Service ( service ) ĊĠĠĠĠĠĠĠ Ġnon Direct Grpc Server Rule . service Registry . add Service ( service ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ// ĠTODO ( mar co ) ĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` Server Ġreceive Ġsus p ends Ġuntil Ġclient Ġinvokes Ġsend ` (){ ĊĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` Client Ġsend Ġsus p ends Ġuntil Ġserver Ġinvokes Ġreceive `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġdeferred Server Channel Ġ= ĠComp letable Deferred < Receive Channel < Hello Request >>() ĊĊĠĠĠĠĠĠĠ Ġsetup Up Server Handler Ġ{ Ġrequest Channel Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdeferred Server Channel . complete ( spy k ( request Channel )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( Long . MAX _ VALUE ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠHello Reply . getDefaultInstance () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġrpc Spy Ġ= ĠRpc Spy () ĊĠĠĠĠĠĠĠ Ġval Ġstub Ġ= Ġrpc Spy . stub ĊĠĠĠĠĠĠĠ Ġval Ġrequest Count Ġ= ĠAtomicInteger () ĊĊĠĠĠĠĠĠĠ Ġassert Fails < Cancellation Exception > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġ( client Request Channel , Ġ_) Ġ= Ġstub ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Coroutine Context ( coroutine Context Ġ+ ĠDispatch ers . Default ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. client Call Client Streaming ( method Descriptor ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlaunch ( Dispatch ers . Default , Ġstart Ġ= ĠCor outine Start . UN DISPATCH ED ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrepeat ( 1 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġclient Request Channel . send ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠHello Request . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. setName ( it . toString ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequest Count . increment AndGet () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġserver Request Channel Ġ= Ġdeferred Server Channel . await () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrepeat ( 3 ){ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 1 0 L ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( it Ġ+ Ġ 1 , Ġrequest Count . get ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġserver Request Channel . receive () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcancel () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġverify ( exact ly Ġ= Ġ 4 ) Ġ{ Ġrpc Spy . call . sendMessage ( any ()) Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` Call Ġcompleted Ġsuccessfully `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġdeferred Server Channel Ġ= ĠComp letable Deferred < Receive Channel < Hello Request >>() ĊĠĠĠĠĠĠĠ Ġval Ġserver Job Ġ= ĠComp letable Deferred < Job >() ĊĠĠĠĠĠĠĠ Ġsetup Up Server Handler Ġ{ Ġrequest Channel Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġjob Ġ= Ġcoroutine Context [ Job ] !! ĊĠĠĠĠĠĠĠĠĠĠĠ Ġjob . invoke On Completion Ġ{ Ġserver Job . complete ( job ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdeferred Server Channel . complete ( spy k ( request Channel )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġreq Values Ġ= Ġrequest Channel . consume As Flow (). map Ġ{ Ġit . name Ġ}. toList () ĊĠĠĠĠĠĠĠĠĠĠĠ ĠHello Reply . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. setMessage ( req Values . join ToString ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġrpc Spy Ġ= ĠRpc Spy () ĊĠĠĠĠĠĠĠ Ġval Ġstub Ġ= Ġrpc Spy . stub ĊĊĠĠĠĠĠĠĠ Ġval Ġ( request Channel , Ġresponse ) Ġ= Ġstub ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. client Call Client Streaming ( method Descriptor ) ĊĊĠĠĠĠĠĠĠ Ġval Ġresponse Value Ġ= Ġrun Blocking ( Dispatch ers . Default ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrepeat ( 3 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequest Channel . send ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠHello Request . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. setName ( it . toString ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrequest Channel . close () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresponse . await () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġverify ( exact ly Ġ= Ġ 0 ) Ġ{ Ġrpc Spy . call . cancel ( any (), Ġany ()) Ġ} ĊĠĠĠĠĠĠĠ Ġassert ( request Channel . is Closed For Send ) Ġ{ Ġ" Request Ġchannel Ġshould Ġbe Ġclosed Ġfor Ġsend " Ġ} ĊĠĠĠĠĠĠĠ ĠassertEquals (" 0 , Ġ 1 , Ġ 2 ", Ġresponse Value . message ) ĊĠĠĠĠĠĠĠ ĠassertFalse ( run Blocking Ġ{ server Job . await () Ġ}. is Cancelled ," Server Ġjob Ġshould Ġnot Ġbe Ġcancelled ") ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` Call Ġis Ġcancelled Ġwhen Ġclient Ġcloses Ġrequest Ġchannel `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġdeferred Server Channel Ġ= ĠComp letable Deferred < Receive Channel < Hello Request >>() ĊĠĠĠĠĠĠĠ Ġval Ġserver Job Ġ= ĠComp letable Deferred < Job >() ĊĠĠĠĠĠĠĠ Ġsetup Up Server Handler Ġ{ Ġrequest Channel Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġjob Ġ= Ġcoroutine Context [ Job ] !! ĊĠĠĠĠĠĠĠĠĠĠĠ Ġjob . invoke On Completion Ġ{ Ġserver Job . complete ( job ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdeferred Server Channel . complete ( spy k ( request Channel )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( Long . MAX _ VALUE ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠHello Reply . getDefaultInstance () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġrpc Spy Ġ= ĠRpc Spy ( non Direct Grpc Server Rule . channel ) ĊĠĠĠĠĠĠĠ Ġval Ġstub Ġ= Ġrpc Spy . stub ĊĠĠĠĠĠĠĠ Ġval Ġexpected Cancel Message Ġ= Ġ" Cancelled Ġby Ġclient Ġwith ĠStream Observer . onError ()" ĊĠĠĠĠĠĠĠ Ġval Ġexpected Exception Ġ= ĠIllegalStateException (" test ") ĊĊĠĠĠĠĠĠĠ Ġval Ġ( request Channel , Ġresponse ) Ġ= Ġstub ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. client Call Client Streaming ( method Descriptor ) ĊĊĠĠĠĠĠĠĠ Ġrun Test Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrequest Channel . send ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠHello Request . newBuilder () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. setName ( 0 . toString ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrequest Channel . close ( expected Exception ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġassert Fails With Status 2 ( Status . CANCEL LED ){ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresponse . await () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġcall State Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġblock Until Cancellation () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġclient . closed . assert Blocking Ġ{ Ġ" Client Ġmust Ġbe Ġclosed " Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġverify ( exact ly Ġ= Ġ 1 ) Ġ{ Ġrpc Spy . call . cancel ( expected Cancel Message , Ġmatch Throwable ( expected Exception )) Ġ} ĊĠĠĠĠĠĠĠ Ġassert ( request Channel . is Closed For Send ) Ġ{ Ġ" Request Ġchannel Ġshould Ġbe Ġclosed Ġfor Ġsend " Ġ} ĊĠĠĠĠĠĠĠ Ġassert Ex Equals ( expected Exception , Ġresponse . get Completion Exception OrNull ()?. cause ) ĊĠĠĠĠĠĠĠ Ġassert ( response . is Cancelled ) Ġ{ Ġ" Response Ġshould Ġnot Ġbe Ġcancelled " Ġ} ĊĠĠĠĠĠĠĠ Ġassert ( run Blocking Ġ{ server Job . await () Ġ}. is Cancelled ){ Ġ" Server Ġjob Ġshould Ġbe Ġcancelled " Ġ} ĊĠĠĠ Ġ} Ċ Ċ } Ċ