import ĠAddress Constants . REMOTE _ AGENT _ ADDRESS _ PREFIX Ċ import ĠAddress Constants . REMOTE _ AGENT _ PROCESS CHAIN LOG S _ SUFFIX Ċ import Ġagent . Local Agent Ċ import Ġagent . Remote Agent Registry Ċ import Ġdb . Submission Registry Ċ import Ġdb . Submission Registry Factory Ċ import Ġhelper . Shell Ċ import Ġhelper . Unique ID Ċ import Ġio . mock k . co Every Ċ import Ġio . mock k . co Verify Ċ import Ġio . mock k . every Ċ import Ġio . mock k . mock k Ċ import Ġio . mock k . mock k Constructor Ċ import Ġio . mock k . mock k Object Ċ import Ġio . mock k . un mock k All Ċ import Ġio . vertx . core . Vert x Ċ import Ġio . vertx . core . event bus . Message Ċ import Ġio . vertx . core . json . JsonObject Ċ import Ġio . vertx . ext . web . handler . impl . HttpStatus Exception Ċ import Ġio . vertx . junit 5 . Vert x Extension Ċ import Ġio . vertx . junit 5 . Vert x TestContext Ċ import Ġio . vertx . kotlin . core . deployment Options Of Ċ import Ġio . vertx . kotlin . core . json . array Ċ import Ġio . vertx . kotlin . core . json . json Ċ import Ġio . vertx . kotlin . core . json . obj Ċ import Ġio . vertx . kotlin . coroutines . dispatcher Ċ import Ġio . vertx . kotlin . coroutines . to Channel Ċ import Ġkotlinx . coroutines . Global Scope Ċ import Ġkotlinx . coroutines . async Ċ import Ġkotlinx . coroutines . channels . Receive Channel Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . launch Ċ import Ġmodel . process chain . Process Chain Ċ import Ġorg . assertj . core . api . Assertions . assertThat Ċ import Ġorg . junit . jupiter . api . After Each Ċ import Ġorg . junit . jupiter . api . BeforeEach Ċ import Ġorg . junit . jupiter . api . Test Ċ import Ġorg . junit . jupiter . api . extension . Extend With Ċ import Ġorg . junit . jupiter . api . io . TempDir Ċ import Ġjava . io . File Ċ import Ġjava . io . IOException Ċ import Ġjava . nio . file . Path Ċ import Ġjava . rmi . Remote Exception Ċ Ċ /** Ċ Ġ* ĠTests Ġfor Ġ[ St eep ] Ċ Ġ* Ġ@ author ĠM ic hel ĠK ra em er Ċ Ġ*/ Ċ @ Extend With ( Vert x Extension :: class ) Ċ class ĠSte ep Test Ġ{ ĊĠ Ġprivate Ġlateinit Ġvar Ġsubmission Registry : ĠSubmission Registry ĊĠ Ġprivate Ġlateinit Ġvar Ġagent Id : ĠString ĊĠ Ġprivate Ġlateinit Ġvar Ġprocess Chain Log Path : ĠString ĊĊĠ Ġ@ BeforeEach ĊĠ Ġfun ĠsetUp ( vertx : ĠVert x , Ġctx : ĠVert x TestContext , Ġ@ TempDir Ġtemp Dir : ĠPath ) Ġ{ ĊĠĠĠ Ġagent Id Ġ= ĠUnique ID . next () ĊĊĠĠĠ Ġ// Ġmock Ġsubmission Ġregistry ĊĠĠĠ Ġsubmission Registry Ġ= Ġmock k () ĊĠĠĠ Ġmock k Object ( Submission Registry Factory ) ĊĠĠĠ Ġevery Ġ{ ĠSubmission Registry Factory . create ( any ()) Ġ} Ġreturns Ġsubmission Registry ĊĊĠĠĠ Ġval Ġprocess Chain Log Path File Ġ= ĠFile ( temp Dir . to File (), Ġ" process chains ") ĊĠĠĠ Ġprocess Chain Log Path File . mkdir s () ĊĠĠĠ Ġprocess Chain Log Path Ġ= Ġprocess Chain Log Path File . absolute Path ĊĊĠĠĠ Ġ// Ġdeploy Ġvert icle Ġunder Ġtest ĊĠĠĠ Ġval Ġconfig Ġ= Ġjson Ġ{ ĊĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠ ĠConfig Constants . AGENT _ CAP AB IL TI IES Ġto Ġarray (" docker ", Ġ" gpu "), ĊĠĠĠĠĠĠĠĠĠ ĠConfig Constants . AGENT _ BUSY _ TIMEOUT Ġto Ġ 1 L , ĊĠĠĠĠĠĠĠĠĠ ĠConfig Constants . AGENT _ ID Ġto Ġagent Id , ĊĠĠĠĠĠĠĠĠĠ ĠConfig Constants . LOG S _ PROCESS CHAIN S _ ENABLED Ġto Ġtrue , ĊĠĠĠĠĠĠĠĠĠ ĠConfig Constants . LOG S _ PROCESS CHAIN S _ PATH Ġto Ġprocess Chain Log Path ĊĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĠĠĠ Ġval Ġoptions Ġ= Ġdeployment Options Of ( config ) ĊĠĠĠ Ġvert x . deploy Vert icle ( St eep :: class . qualified Name , Ġoptions , Ġctx . comp leting ()) ĊĠ Ġ} ĊĊĠ Ġ@ After Each ĊĠ Ġfun ĠtearDown () Ġ{ ĊĠĠĠ Ġun mock k All () ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠExecute Ġan Ġempty Ġprocess Ġchain ĊĠĠ Ġ*/ ĊĠ Ġ@ Test ĊĠ Ġfun Ġexecute Empty Process Chain ( vertx : ĠVert x , Ġctx : ĠVert x TestContext ) Ġ{ ĊĠĠĠ Ġval Ġprocess Chain Ġ= ĠProcess Chain () ĊĠĠĠ Ġval Ġremote Agent Registry Ġ= ĠRemote Agent Registry ( vertx ) ĊĊĠĠĠ ĠGlobal Scope . launch ( vertx . dispatcher ()) Ġ{ ĊĠĠĠĠĠ Ġval Ġcandidates Ġ= Ġremote Agent Registry . select Candidates ( ĊĠĠĠĠĠĠĠĠĠ ĠlistOf ( process Chain . required Capabilities Ġto Ġ 1 )) ĊĠĠĠĠĠ Ġctx . co Verify Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġagent Ġ= Ġremote Agent Registry . try Allocate ( candidates [ 0 ]. second , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprocess Chain . id ) ĊĠĠĠĠĠĠĠ ĠassertThat ( agent ). isNotNull ĊĊĠĠĠĠĠĠĠ Ġval Ġresults Ġ= Ġagent !!. execute ( process Chain ) ĊĠĠĠĠĠĠĠ ĠassertThat ( results ). isEmpty () ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġctx . complete Now () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠExecute Ġa Ġsimple Ġprocess Ġchain ĊĠĠ Ġ*/ ĊĠ Ġ@ Test ĊĠ Ġfun Ġexecute Process Chain ( vertx : ĠVert x , Ġctx : ĠVert x TestContext ) Ġ{ ĊĠĠĠ Ġval Ġprocess Chain Ġ= ĠProcess Chain () ĊĠĠĠ Ġval Ġremote Agent Registry Ġ= ĠRemote Agent Registry ( vertx ) ĊĠĠĠ Ġval Ġexpected Results Ġ= Ġmap Of (" output _ files " Ġto ĠlistOf (" test 1 ", Ġ" test 2 ")) ĊĊĠĠĠ Ġmock k Constructor ( Local Agent :: class ) ĊĠĠĠ Ġco Every Ġ{ Ġany Construct ed < Local Agent >(). execute ( process Chain ) Ġ} Ġreturns Ġexpected Results ĊĊĠĠĠ ĠGlobal Scope . launch ( vertx . dispatcher ()) Ġ{ ĊĠĠĠĠĠ Ġval Ġcandidates Ġ= Ġremote Agent Registry . select Candidates ( ĊĠĠĠĠĠĠĠĠĠ ĠlistOf ( process Chain . required Capabilities Ġto Ġ 1 )) ĊĠĠĠĠĠ Ġctx . co Verify Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġagent Ġ= Ġremote Agent Registry . try Allocate ( candidates [ 0 ]. second , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprocess Chain . id ) ĊĠĠĠĠĠĠĠ ĠassertThat ( agent ). isNotNull ĊĊĠĠĠĠĠĠĠ Ġval Ġresults Ġ= Ġagent !!. execute ( process Chain ) ĊĠĠĠĠĠĠĠ ĠassertThat ( results ). isEqualTo ( expected Results ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġctx . complete Now () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠTest Ġwhat Ġhappens Ġif Ġ[ Local Agent ] Ġthrows Ġan Ġexception ĊĠĠ Ġ*/ ĊĠ Ġ@ Test ĊĠ Ġfun Ġerror In Local Agent ( vertx : ĠVert x , Ġctx : ĠVert x TestContext ) Ġ{ ĊĠĠĠ Ġval Ġprocess Chain Ġ= ĠProcess Chain () ĊĠĠĠ Ġval Ġremote Agent Registry Ġ= ĠRemote Agent Registry ( vertx ) ĊĠĠĠ Ġval ĠerrorMessage Ġ= Ġ" File Ġnot Ġfound " ĊĊĠĠĠ Ġmock k Constructor ( Local Agent :: class ) ĊĠĠĠ Ġco Every Ġ{ Ġany Construct ed < Local Agent >(). execute ( process Chain ) Ġ} Ġthrows ĊĠĠĠĠĠĠĠ ĠIOException ( errorMessage ) ĊĊĠĠĠ ĠGlobal Scope . launch ( vertx . dispatcher ()) Ġ{ ĊĠĠĠĠĠ Ġval Ġcandidates Ġ= Ġremote Agent Registry . select Candidates ( ĊĠĠĠĠĠĠĠĠĠ ĠlistOf ( process Chain . required Capabilities Ġto Ġ 1 )) ĊĠĠĠĠĠ Ġctx . co Verify Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġagent Ġ= Ġremote Agent Registry . try Allocate ( candidates [ 0 ]. second , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprocess Chain . id ) ĊĠĠĠĠĠĠĠ ĠassertThat Thrown By Ġ{ Ġagent !!. execute ( process Chain ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. isInstanceOf ( Remote Exception :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. has Message ( errorMessage ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġctx . complete Now () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠTest Ġwhat Ġhappens Ġif Ġ[ Local Agent ] Ġthrows Ġa Ġ[ Shell . ExecutionException ] ĊĠĠ Ġ*/ ĊĠ Ġ@ Test ĊĠ Ġfun Ġexecution Exception In Local Agent ( vertx : ĠVert x , Ġctx : ĠVert x TestContext ) Ġ{ ĊĠĠĠ Ġval Ġprocess Chain Ġ= ĠProcess Chain () ĊĠĠĠ Ġval Ġremote Agent Registry Ġ= ĠRemote Agent Registry ( vertx ) ĊĠĠĠ Ġval ĠerrorMessage Ġ= Ġ" Could Ġnot Ġgenerate Ġfile " ĊĠĠĠ Ġval Ġlast Output Ġ= Ġ" This Ġis Ġthe Ġlast Ġoutput " ĊĠĠĠ Ġval Ġexit Code Ġ= Ġ 1 3 2 ĊĊĠĠĠ Ġmock k Constructor ( Local Agent :: class ) ĊĠĠĠ Ġco Every Ġ{ Ġany Construct ed < Local Agent >(). execute ( process Chain ) Ġ} Ġthrows ĊĠĠĠĠĠĠĠ ĠShell . ExecutionException ( errorMessage , Ġlast Output , Ġexit Code ) ĊĊĠĠĠ ĠGlobal Scope . launch ( vertx . dispatcher ()) Ġ{ ĊĠĠĠĠĠ Ġval Ġcandidates Ġ= Ġremote Agent Registry . select Candidates ( ĊĠĠĠĠĠĠĠĠĠ ĠlistOf ( process Chain . required Capabilities Ġto Ġ 1 )) ĊĠĠĠĠĠ Ġctx . co Verify Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġagent Ġ= Ġremote Agent Registry . try Allocate ( candidates [ 0 ]. second , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprocess Chain . id ) ĊĠĠĠĠĠĠĠ ĠassertThat ( agent ). isNotNull ĊĠĠĠĠĠĠĠ ĠassertThat Thrown By Ġ{ Ġagent !!. execute ( process Chain ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. isInstanceOf ( Remote Exception :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. has Message ("$ errorMessage \ n \ n Exit Ġcode : Ġ$ exit Code \ n \ n $ last Output ") ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġctx . complete Now () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠExecute Ġa Ġsimple Ġprocess Ġchain Ġand Ġcheck Ġif Ġthe Ġoperation Ġis Ġide mpot ent ĊĠĠ Ġ* Ġ( i . e . Ġif Ġwe Ġcan Ġcall Ġ[ agent . Remote Agent . execute ] Ġmultiple Ġtimes Ġbut Ġstill ĊĠĠ Ġ* Ġonly Ġexecute Ġthe Ġprocess Ġchain Ġonce ) ĊĠĠ Ġ*/ ĊĠ Ġ@ Test ĊĠ Ġfun Ġexecute Process Chain Ide mpot ent ( vertx : ĠVert x , Ġctx : ĠVert x TestContext ) Ġ{ ĊĠĠĠ Ġval Ġprocess Chain Ġ= ĠProcess Chain () ĊĠĠĠ Ġval Ġremote Agent Registry Ġ= ĠRemote Agent Registry ( vertx ) ĊĠĠĠ Ġval Ġexpected Results Ġ= Ġmap Of (" output _ files " Ġto ĠlistOf (" test 1 ", Ġ" test 2 ")) ĊĊĠĠĠ Ġmock k Constructor ( Local Agent :: class ) ĊĠĠĠ Ġco Every Ġ{ Ġany Construct ed < Local Agent >(). execute ( process Chain ) Ġ} Ġco Answers Ġ{ ĊĠĠĠĠĠ Ġ// Ġpret end Ġit Ġtakes Ġ 1 s Ġto Ġexecute Ġthe Ġprocess Ġchain ĊĠĠĠĠĠ Ġdelay ( 1 0 0 0 ) ĊĠĠĠĠĠ Ġexpected Results ĊĠĠĠ Ġ} ĊĊĠĠĠ ĠGlobal Scope . launch ( vertx . dispatcher ()) Ġ{ ĊĠĠĠĠĠ Ġval Ġcandidates Ġ= Ġremote Agent Registry . select Candidates ( ĊĠĠĠĠĠĠĠĠĠ ĠlistOf ( process Chain . required Capabilities Ġto Ġ 1 )) ĊĠĠĠĠĠ Ġctx . co Verify Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġagent Ġ= Ġremote Agent Registry . try Allocate ( candidates [ 0 ]. second , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprocess Chain . id ) ĊĠĠĠĠĠĠĠ ĠassertThat ( agent ). isNotNull ĊĊĠĠĠĠĠĠĠ Ġval Ġd 1 Ġ= Ġasync Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġagent !!. execute ( process Chain ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġd 2 Ġ= Ġasync Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġagent !!. execute ( process Chain ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġresults 1 Ġ= Ġd 1 . await () ĊĠĠĠĠĠĠĠ Ġval Ġresults 2 Ġ= Ġd 2 . await () ĊĊĠĠĠĠĠĠĠ ĠassertThat ( results 1 ). isEqualTo ( expected Results ) ĊĠĠĠĠĠĠĠ ĠassertThat ( results 2 ). isEqualTo ( expected Results ) ĊĊĠĠĠĠĠĠĠ Ġ// Ġmake Ġsure Ġ` execute ` Ġwas Ġonly Ġcalled Ġonce ĊĠĠĠĠĠĠĠ Ġco Verify ( exact ly Ġ= Ġ 1 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġany Construct ed < Local Agent >(). execute ( process Chain ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠ Ġctx . complete Now () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠExecute Ġa Ġsimple Ġprocess Ġchain Ġand Ġcheck Ġif Ġthe Ġoperation Ġis Ġide mpot ent ĊĠĠ Ġ* Ġeven Ġif Ġwe Ġallocate Ġthe Ġagent Ġtwice Ġand Ġhave Ġtwo Ġdifferent ĊĠĠ Ġ* Ġ[ agent . Remote Agent ] Ġinstances ĊĠĠ Ġ*/ ĊĠ Ġ@ Test ĊĠ Ġfun Ġexecute Process Chain Allocate Ide mpot ent ( vertx : ĠVert x , Ġctx : ĠVert x TestContext ) Ġ{ ĊĠĠĠ Ġval Ġprocess Chain Ġ= ĠProcess Chain () ĊĠĠĠ Ġval Ġremote Agent Registry Ġ= ĠRemote Agent Registry ( vertx ) ĊĠĠĠ Ġval Ġexpected Results Ġ= Ġmap Of (" output _ files " Ġto ĠlistOf (" test 1 ", Ġ" test 2 ")) ĊĊĠĠĠ Ġmock k Constructor ( Local Agent :: class ) ĊĠĠĠ Ġco Every Ġ{ Ġany Construct ed < Local Agent >(). execute ( process Chain ) Ġ} Ġco Answers Ġ{ ĊĠĠĠĠĠ Ġ// Ġpret end Ġit Ġtakes Ġ 1 s Ġto Ġexecute Ġthe Ġprocess Ġchain ĊĠĠĠĠĠ Ġdelay ( 1 0 0 0 ) ĊĠĠĠĠĠ Ġexpected Results ĊĠĠĠ Ġ} ĊĊĠĠĠ ĠGlobal Scope . launch ( vertx . dispatcher ()) Ġ{ ĊĠĠĠĠĠ Ġval Ġcandidates Ġ= Ġremote Agent Registry . select Candidates ( ĊĠĠĠĠĠĠĠĠĠ ĠlistOf ( process Chain . required Capabilities Ġto Ġ 1 )) ĊĠĠĠĠĠ Ġctx . co Verify Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġagent 1 Ġ= Ġremote Agent Registry . try Allocate ( candidates [ 0 ]. second , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprocess Chain . id ) ĊĠĠĠĠĠĠĠ ĠassertThat ( agent 1 ). isNotNull ĊĊĠĠĠĠĠĠĠ Ġval Ġd 1 Ġ= Ġasync Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġagent 1 !!. execute ( process Chain ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġdelay ( 5 0 0 ) ĊĊĠĠĠĠĠĠĠ Ġval Ġagent 2 Ġ= Ġremote Agent Registry . try Allocate ( candidates [ 0 ]. second , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprocess Chain . id ) ĊĠĠĠĠĠĠĠ ĠassertThat ( agent 2 ). isNotNull ĊĠĠĠĠĠĠĠ Ġval Ġd 2 Ġ= Ġasync Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġagent 2 !!. execute ( process Chain ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġresults 1 Ġ= Ġd 1 . await () ĊĠĠĠĠĠĠĠ Ġval Ġresults 2 Ġ= Ġd 2 . await () ĊĊĠĠĠĠĠĠĠ ĠassertThat ( results 1 ). isEqualTo ( expected Results ) ĊĠĠĠĠĠĠĠ ĠassertThat ( results 2 ). isEqualTo ( expected Results ) ĊĊĠĠĠĠĠĠĠ Ġ// Ġmake Ġsure Ġ` execute ` Ġwas Ġonly Ġcalled Ġonce ĊĠĠĠĠĠĠĠ Ġco Verify ( exact ly Ġ= Ġ 1 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġany Construct ed < Local Agent >(). execute ( process Chain ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠ Ġctx . complete Now () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠTest Ġwhat Ġhappens Ġif Ġthe Ġagent Ġdoes Ġnot Ġhave Ġthe Ġrequired Ġcapabilities ĊĠĠ Ġ*/ ĊĠ Ġ@ Test ĊĠ Ġfun Ġwrong Capabilities ( vertx : ĠVert x , Ġctx : ĠVert x TestContext ) Ġ{ ĊĠĠĠ Ġval Ġprocess Chain Ġ= ĠProcess Chain ( required Capabilities Ġ= Ġset Of (" docker ", Ġ" foobar ")) ĊĠĠĠ Ġval Ġremote Agent Registry Ġ= ĠRemote Agent Registry ( vertx ) ĊĊĠĠĠ ĠGlobal Scope . launch ( vertx . dispatcher ()) Ġ{ ĊĠĠĠĠĠ Ġval Ġcandidates Ġ= Ġremote Agent Registry . select Candidates ( ĊĠĠĠĠĠĠĠĠĠ ĠlistOf ( process Chain . required Capabilities Ġto Ġ 1 )) ĊĠĠĠĠĠ Ġctx . verify Ġ{ ĊĠĠĠĠĠĠĠ ĠassertThat ( candidates ). isEmpty () ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġctx . complete Now () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠTest Ġif Ġwe Ġare Ġavailable Ġif Ġthe Ġrequired Ġcapabilities Ġmatch ĊĠĠ Ġ*/ ĊĠ Ġ@ Test ĊĠ Ġfun Ġright Capabilities ( vertx : ĠVert x , Ġctx : ĠVert x TestContext ) Ġ{ ĊĠĠĠ Ġval Ġprocess Chain Ġ= ĠProcess Chain ( required Capabilities Ġ= Ġset Of (" docker ")) ĊĠĠĠ Ġval Ġremote Agent Registry Ġ= ĠRemote Agent Registry ( vertx ) ĊĊĠĠĠ ĠGlobal Scope . launch ( vertx . dispatcher ()) Ġ{ ĊĠĠĠĠĠ Ġval Ġcandidates Ġ= Ġremote Agent Registry . select Candidates ( ĊĠĠĠĠĠĠĠĠĠ ĠlistOf ( process Chain . required Capabilities Ġto Ġ 1 )) ĊĠĠĠĠĠ Ġctx . verify Ġ{ ĊĠĠĠĠĠĠĠ ĠassertThat ( candidates ). contains Exactly ( Pair ( process Chain . required Capabilities , ĊĠĠĠĠĠĠĠĠĠĠĠ ĠRE MOTE _ AGENT _ ADDRESS _ PREFIX Ġ+ Ġagent Id )) ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġctx . complete Now () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠTest Ġif Ġthe Ġagent Ġselects Ġthe Ġbest Ġrequired Ġcapabilities Ġby Ġmaximum Ġcount ĊĠĠ Ġ* Ġof Ġcorresponding Ġprocess Ġchains ĊĠĠ Ġ*/ ĊĠ Ġ@ Test ĊĠ Ġfun Ġbest Capabilities By Count ( vertx : ĠVert x , Ġctx : ĠVert x TestContext ) Ġ{ ĊĠĠĠ Ġval Ġrequired Capabilities 1 Ġ= Ġset Of (" docker ") ĊĠĠĠ Ġval Ġrequired Capabilities 2 Ġ= Ġset Of (" gpu ") ĊĠĠĠ Ġval Ġremote Agent Registry Ġ= ĠRemote Agent Registry ( vertx ) ĊĊĠĠĠ ĠGlobal Scope . launch ( vertx . dispatcher ()) Ġ{ ĊĠĠĠĠĠ Ġval Ġcandidates 1 Ġ= Ġremote Agent Registry . select Candidates ( ĊĠĠĠĠĠĠĠĠĠ ĠlistOf ( required Capabilities 1 Ġto Ġ 1 , Ġrequired Capabilities 2 Ġto Ġ 2 )) ĊĠĠĠĠĠ Ġval Ġcandidates 2 Ġ= Ġremote Agent Registry . select Candidates ( ĊĠĠĠĠĠĠĠĠĠ ĠlistOf ( required Capabilities 1 Ġto Ġ 2 , Ġrequired Capabilities 2 Ġto Ġ 1 )) ĊĠĠĠĠĠ Ġval Ġcandidates 3 Ġ= Ġremote Agent Registry . select Candidates ( ĊĠĠĠĠĠĠĠĠĠ ĠlistOf ( required Capabilities 1 Ġto Ġ 1 , Ġrequired Capabilities 2 Ġto Ġ 1 )) ĊĠĠĠĠĠ Ġctx . verify Ġ{ ĊĠĠĠĠĠĠĠ ĠassertThat ( candidates 1 ). contains Exactly ( Pair ( required Capabilities 2 , ĊĠĠĠĠĠĠĠĠĠĠĠ ĠRE MOTE _ AGENT _ ADDRESS _ PREFIX Ġ+ Ġagent Id )) ĊĠĠĠĠĠĠĠ ĠassertThat ( candidates 2 ). contains Exactly ( Pair ( required Capabilities 1 , ĊĠĠĠĠĠĠĠĠĠĠĠ ĠRE MOTE _ AGENT _ ADDRESS _ PREFIX Ġ+ Ġagent Id )) ĊĠĠĠĠĠĠĠ ĠassertThat ( candidates 3 ). contains Exactly ( Pair ( required Capabilities 1 , ĊĠĠĠĠĠĠĠĠĠĠĠ ĠRE MOTE _ AGENT _ ADDRESS _ PREFIX Ġ+ Ġagent Id )) ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġctx . complete Now () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠTest Ġif Ġwe Ġcan Ġbe Ġallocated Ġand Ġde allocated ĊĠĠ Ġ*/ ĊĠ Ġ@ Test ĊĠ Ġfun Ġallocate De allocate ( vertx : ĠVert x , Ġctx : ĠVert x TestContext ) Ġ{ ĊĠĠĠ Ġval Ġremote Agent Registry Ġ= ĠRemote Agent Registry ( vertx ) ĊĊĠĠĠ ĠGlobal Scope . launch ( vertx . dispatcher ()) Ġ{ ĊĠĠĠĠĠ Ġval Ġcandidates Ġ= Ġremote Agent Registry . select Candidates ( ĊĠĠĠĠĠĠĠĠĠ ĠlistOf ( empty Set < String >() Ġto Ġ 1 )) ĊĠĠĠĠĠ Ġctx . co Verify Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġagent 1 Ġ= Ġremote Agent Registry . try Allocate ( candidates [ 0 ]. second , ĊĠĠĠĠĠĠĠĠĠĠĠ ĠUnique ID . next ()) ĊĠĠĠĠĠĠĠ ĠassertThat ( agent 1 ). isNotNull ĊĊĠĠĠĠĠĠĠ Ġval Ġagent 2 Ġ= Ġremote Agent Registry . try Allocate ( candidates [ 0 ]. second , ĊĠĠĠĠĠĠĠĠĠĠĠ ĠUnique ID . next ()) ĊĠĠĠĠĠĠĠ ĠassertThat ( agent 2 ). isNull () ĊĊĠĠĠĠĠĠĠ Ġremote Agent Registry . de allocate ( agent 1 !! ) ĊĠĠĠĠĠĠĠ Ġval Ġagent 3 Ġ= Ġremote Agent Registry . try Allocate ( candidates [ 0 ]. second , ĊĠĠĠĠĠĠĠĠĠĠĠ ĠUnique ID . next ()) ĊĠĠĠĠĠĠĠ ĠassertThat ( agent 3 ). isNotNull ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġctx . complete Now () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠTest Ġif Ġthe Ġagent Ġbecomes Ġavailable Ġagain Ġafter Ġnot Ġhaving Ġreceived Ġa ĊĠĠ Ġ* Ġprocess Ġchain Ġfor Ġtoo Ġlong ĊĠĠ Ġ*/ ĊĠ Ġ@ Test ĊĠ Ġfun Ġidle ( vertx : ĠVert x , Ġctx : ĠVert x TestContext ) Ġ{ ĊĠĠĠ Ġval Ġremote Agent Registry Ġ= ĠRemote Agent Registry ( vertx ) ĊĊĠĠĠ ĠGlobal Scope . launch ( vertx . dispatcher ()) Ġ{ ĊĠĠĠĠĠ Ġval Ġcandidates Ġ= Ġremote Agent Registry . select Candidates ( ĊĠĠĠĠĠĠĠĠĠ ĠlistOf ( empty Set < String >() Ġto Ġ 1 )) ĊĠĠĠĠĠ Ġctx . co Verify Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġagent 1 Ġ= Ġremote Agent Registry . try Allocate ( candidates [ 0 ]. second , ĊĠĠĠĠĠĠĠĠĠĠĠ ĠUnique ID . next ()) ĊĠĠĠĠĠĠĠ ĠassertThat ( agent 1 ). isNotNull ĊĊĠĠĠĠĠĠĠ Ġval Ġagent 2 Ġ= Ġremote Agent Registry . try Allocate ( candidates [ 0 ]. second , ĊĠĠĠĠĠĠĠĠĠĠĠ ĠUnique ID . next ()) ĊĠĠĠĠĠĠĠ ĠassertThat ( agent 2 ). isNull () ĊĊĠĠĠĠĠĠĠ Ġdelay ( 1 0 0 1 ) ĊĊĠĠĠĠĠĠĠ Ġval Ġagent 3 Ġ= Ġremote Agent Registry . try Allocate ( candidates [ 0 ]. second , ĊĠĠĠĠĠĠĠĠĠĠĠ ĠUnique ID . next ()) ĊĠĠĠĠĠĠĠ ĠassertThat ( agent 3 ). isNotNull ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġctx . complete Now () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠTest Ġif Ġwe Ġcan Ġre allocate Ġan Ġagent Ġif Ġwe Ġsend Ġthe Ġsame Ġprocess Ġchain ĠID Ġagain ĊĠĠ Ġ*/ ĊĠ Ġ@ Test ĊĠ Ġfun Ġide mpot ent Allocate ( vertx : ĠVert x , Ġctx : ĠVert x TestContext ) Ġ{ ĊĠĠĠ Ġval Ġprocess Chain Ġ= ĠProcess Chain () ĊĠĠĠ Ġval Ġremote Agent Registry Ġ= ĠRemote Agent Registry ( vertx ) ĊĊĠĠĠ ĠGlobal Scope . launch ( vertx . dispatcher ()) Ġ{ ĊĠĠĠĠĠ Ġval Ġcandidates Ġ= Ġremote Agent Registry . select Candidates ( ĊĠĠĠĠĠĠĠĠĠ ĠlistOf ( process Chain . required Capabilities Ġto Ġ 1 )) ĊĠĠĠĠĠ Ġctx . co Verify Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġagent 1 Ġ= Ġremote Agent Registry . try Allocate ( candidates [ 0 ]. second , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprocess Chain . id ) ĊĠĠĠĠĠĠĠ ĠassertThat ( agent 1 ). isNotNull ĊĊĠĠĠĠĠĠĠ Ġval Ġagent 2 Ġ= Ġremote Agent Registry . try Allocate ( candidates [ 0 ]. second , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprocess Chain . id ) ĊĠĠĠĠĠĠĠ ĠassertThat ( agent 2 ). isNotNull ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġctx . complete Now () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠTest Ġif Ġthe Ġagent Ġbecomes Ġavailable Ġagain Ġafter Ġbeing Ġidle Ġfor Ġtoo Ġlong ĊĠĠ Ġ*/ ĊĠ Ġ@ Test ĊĠ Ġfun Ġidle After Execute ( vertx : ĠVert x , Ġctx : ĠVert x TestContext ) Ġ{ ĊĠĠĠ Ġval Ġprocess Chain Ġ= ĠProcess Chain () ĊĠĠĠ Ġval Ġremote Agent Registry Ġ= ĠRemote Agent Registry ( vertx ) ĊĊĠĠĠ Ġmock k Constructor ( Local Agent :: class ) ĊĠĠĠ Ġco Every Ġ{ Ġany Construct ed < Local Agent >(). execute ( process Chain ) Ġ} Ġreturns Ġempty Map () ĊĊĠĠĠ ĠGlobal Scope . launch ( vertx . dispatcher ()) Ġ{ ĊĠĠĠĠĠ Ġval Ġcandidates Ġ= Ġremote Agent Registry . select Candidates ( ĊĠĠĠĠĠĠĠĠĠ ĠlistOf ( process Chain . required Capabilities Ġto Ġ 1 )) ĊĠĠĠĠĠ Ġctx . co Verify Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġagent Ġ= Ġremote Agent Registry . try Allocate ( candidates [ 0 ]. second , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprocess Chain . id ) ĊĠĠĠĠĠĠĠ ĠassertThat ( agent ). isNotNull ĊĠĠĠĠĠĠĠ Ġagent !!. execute ( process Chain ) ĊĊĠĠĠĠĠĠĠ Ġdelay ( 1 0 0 1 ) ĊĊĠĠĠĠĠĠĠ Ġval Ġagent 2 Ġ= Ġremote Agent Registry . try Allocate ( candidates [ 0 ]. second , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprocess Chain . id ) ĊĠĠĠĠĠĠĠ ĠassertThat ( agent 2 ). isNotNull ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġctx . complete Now () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġprivate Ġsuspend Ġfun Ġreceive Process Chain LogFile ( channel : ĠReceive Channel < Message < JsonObject >> ): ĊĠĠĠĠĠ ĠPair < String , ĠJsonObject ?> Ġ{ ĊĠĠĠ Ġvar Ġheader : ĠJsonObject ? Ġ= Ġnull ĊĠĠĠ Ġval Ġreceived Contents Ġ= ĠStringBuilder () ĊĠĠĠ Ġfor Ġ( reply Ġin Ġchannel ) Ġ{ ĊĠĠĠĠĠ Ġval Ġobj Ġ= Ġreply . body () ĊĠĠĠĠĠ Ġif Ġ( obj . isEmpty ) Ġ{ ĊĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( obj . getLong (" size ") Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠ Ġheader Ġ= Ġobj ĊĠĠĠĠĠĠĠ Ġreply . reply ( null ) ĊĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( obj . getString (" data ") Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠ Ġreceived Contents . append ( obj . getString (" data ")) ĊĠĠĠĠĠĠĠ Ġreply . reply ( null ) ĊĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( obj . getInteger (" error ") Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠ Ġthrow ĠHttpStatus Exception ( obj . getInteger (" error "), Ġobj . getString (" message ")) ĊĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠ Ġthrow ĠIllegalStateException (" Illegal Ġmessage : Ġ${ obj . encode () }") ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġreturn Ġ( received Contents . toString () Ġto Ġheader ) ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠTest Ġif Ġwe Ġcan Ġfetch Ġa Ġprocess Ġchain Ġlog Ġfile ĊĠĠ Ġ*/ ĊĠ Ġ@ Test ĊĠ Ġfun Ġfetch Process Chain LogFile ( vertx : ĠVert x , Ġctx : ĠVert x TestContext ) Ġ{ ĊĠĠĠ Ġval Ġid Ġ= Ġ" abcdef g 1 2 3 4 5 6 " ĊĠĠĠ Ġval Ġcontents Ġ= Ġ" Hello Ġworld " ĊĠĠĠ Ġval Ġlog File Ġ= ĠFile ( process Chain Log Path , Ġ"$ id . log ") ĊĠĠĠ Ġlog File . write Text ( contents ) ĊĊĠĠĠ Ġval Ġaddress Ġ= ĠRE MOTE _ AGENT _ ADDRESS _ PREFIX Ġ+ Ġagent Id Ġ+ ĊĠĠĠĠĠĠĠ ĠRE MOTE _ AGENT _ PROCESS CHAIN LOG S _ SUFFIX ĊĠĠĠ Ġval Ġreply Address Ġ= Ġ"$ address . reply .${ Unique ID . next () }" ĊĊĠĠĠ ĠGlobal Scope . launch ( vertx . dispatcher ()) Ġ{ ĊĠĠĠĠĠ Ġctx . co Verify Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġconsumer Ġ= Ġvert x . event Bus (). consumer < JsonObject >( reply Address ) ĊĠĠĠĠĠĠĠ Ġval Ġchannel Ġ= Ġconsumer . to Channel ( vertx ) ĊĊĠĠĠĠĠĠĠ Ġval Ġmsg Ġ= Ġjson Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" action " Ġto Ġ" fetch ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" id " Ġto Ġid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" reply Address " Ġto Ġreply Address ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġvert x . event Bus (). send ( address , Ġmsg ) ĊĊĠĠĠĠĠĠĠ Ġval Ġ( received Contents , Ġheader ) Ġ= Ġreceive Process Chain LogFile ( channel ) ĊĊĠĠĠĠĠĠĠ Ġval Ġsize Ġ= Ġheader ?. getLong (" size ") ĊĠĠĠĠĠĠĠ ĠassertThat ( size ). isEqualTo ( contents . length . to Long ()) ĊĠĠĠĠĠĠĠ ĠassertThat ( received Contents ). isEqualTo ( contents ) ĊĊĠĠĠĠĠĠĠ Ġctx . complete Now () ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠTest Ġif Ġwe Ġcan Ġcheck Ġif Ġa Ġprocess Ġchain Ġlog Ġfile Ġexists ĊĠĠ Ġ*/ ĊĠ Ġ@ Test ĊĠ Ġfun Ġexists Process Chain LogFile ( vertx : ĠVert x , Ġctx : ĠVert x TestContext ) Ġ{ ĊĠĠĠ Ġval Ġid Ġ= Ġ" abcdef g 1 2 3 4 5 6 " ĊĠĠĠ Ġval Ġcontents Ġ= Ġ" Hello Ġworld " ĊĠĠĠ Ġval Ġlog File Ġ= ĠFile ( process Chain Log Path , Ġ"$ id . log ") ĊĠĠĠ Ġlog File . write Text ( contents ) ĊĊĠĠĠ Ġval Ġaddress Ġ= ĠRE MOTE _ AGENT _ ADDRESS _ PREFIX Ġ+ Ġagent Id Ġ+ ĊĠĠĠĠĠĠĠ ĠRE MOTE _ AGENT _ PROCESS CHAIN LOG S _ SUFFIX ĊĠĠĠ Ġval Ġreply Address Ġ= Ġ"$ address . reply .${ Unique ID . next () }" ĊĊĠĠĠ ĠGlobal Scope . launch ( vertx . dispatcher ()) Ġ{ ĊĠĠĠĠĠ Ġctx . co Verify Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġconsumer Ġ= Ġvert x . event Bus (). consumer < JsonObject >( reply Address ) ĊĠĠĠĠĠĠĠ Ġval Ġchannel Ġ= Ġconsumer . to Channel ( vertx ) ĊĊĠĠĠĠĠĠĠ Ġval Ġmsg Ġ= Ġjson Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" action " Ġto Ġ" exists ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" id " Ġto Ġid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" reply Address " Ġto Ġreply Address ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġvert x . event Bus (). send ( address , Ġmsg ) ĊĊĠĠĠĠĠĠĠ Ġval Ġobj Ġ= Ġchannel . receive (). body () ĊĠĠĠĠĠĠĠ ĠassertThat ( obj . getInteger (" size ")). isEqualTo ( contents . length . to Long ()) ĊĊĠĠĠĠĠĠĠ Ġctx . complete Now () ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠTest Ġif Ġwe Ġfetching Ġa Ġprocess Ġchain Ġlog Ġfile Ġfails Ġif Ġit Ġdoes Ġnot Ġexist ĊĠĠ Ġ*/ ĊĠ Ġ@ Test ĊĠ Ġfun Ġfetch Non Existing Process Chain LogFile ( vertx : ĠVert x , Ġctx : ĠVert x TestContext ) Ġ{ ĊĠĠĠ Ġval Ġid Ġ= Ġ" FO OB AR " ĊĊĠĠĠ Ġval Ġaddress Ġ= ĠRE MOTE _ AGENT _ ADDRESS _ PREFIX Ġ+ Ġagent Id Ġ+ ĊĠĠĠĠĠĠĠ ĠRE MOTE _ AGENT _ PROCESS CHAIN LOG S _ SUFFIX ĊĠĠĠ Ġval Ġreply Address Ġ= Ġ"$ address . reply .${ Unique ID . next () }" ĊĊĠĠĠ ĠGlobal Scope . launch ( vertx . dispatcher ()) Ġ{ ĊĠĠĠĠĠ Ġctx . co Verify Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġconsumer Ġ= Ġvert x . event Bus (). consumer < JsonObject >( reply Address ) ĊĠĠĠĠĠĠĠ Ġval Ġchannel Ġ= Ġconsumer . to Channel ( vertx ) ĊĊĠĠĠĠĠĠĠ Ġval Ġmsg Ġ= Ġjson Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" action " Ġto Ġ" fetch ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" id " Ġto Ġid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" reply Address " Ġto Ġreply Address ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġvert x . event Bus (). send ( address , Ġmsg ) ĊĊĠĠĠĠĠĠĠ Ġval Ġobj Ġ= Ġchannel . receive (). body () ĊĠĠĠĠĠĠĠ ĠassertThat ( obj . getInteger (" error ")). isEqualTo ( 4 0 4 ) ĊĊĠĠĠĠĠĠĠ Ġctx . complete Now () ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ/** ĊĠĠ Ġ* ĠTest Ġif Ġwe Ġcan Ġpartially Ġfetch Ġa Ġprocess Ġchain Ġlog Ġfile ĊĠĠ Ġ*/ ĊĠ Ġ@ Test ĊĠ Ġfun Ġfetch Process Chain LogFile Partial ( vertx : ĠVert x , Ġctx : ĠVert x TestContext ) Ġ{ ĊĠĠĠ Ġval Ġid Ġ= Ġ" abcdef g 1 2 3 4 5 6 " ĊĠĠĠ Ġval Ġcontents Ġ= Ġ" Hello Ġworld " ĊĠĠĠ Ġval Ġlog File Ġ= ĠFile ( process Chain Log Path , Ġ"$ id . log ") ĊĠĠĠ Ġlog File . write Text ( contents ) ĊĊĠĠĠ Ġval Ġaddress Ġ= ĠRE MOTE _ AGENT _ ADDRESS _ PREFIX Ġ+ Ġagent Id Ġ+ ĊĠĠĠĠĠĠĠ ĠRE MOTE _ AGENT _ PROCESS CHAIN LOG S _ SUFFIX ĊĠĠĠ Ġval Ġreply Address Ġ= Ġ"$ address . reply .${ Unique ID . next () }" ĊĊĠĠĠ ĠGlobal Scope . launch ( vertx . dispatcher ()) Ġ{ ĊĠĠĠĠĠ Ġctx . co Verify Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġconsumer Ġ= Ġvert x . event Bus (). consumer < JsonObject >( reply Address ) ĊĠĠĠĠĠĠĠ Ġval Ġchannel Ġ= Ġconsumer . to Channel ( vertx ) ĊĊĠĠĠĠĠĠĠ Ġ// Ġrequest Ġthe Ġfirst Ġbyte ĊĠĠĠĠĠĠĠ Ġval Ġmsg 1 Ġ= Ġjson Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" action " Ġto Ġ" fetch ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" id " Ġto Ġid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" reply Address " Ġto Ġreply Address , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" end " Ġto Ġ 0 ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġvert x . event Bus (). send ( address , Ġmsg 1 ) ĊĊĠĠĠĠĠĠĠ Ġval Ġ( received Contents 1 , Ġheader 1 ) Ġ= Ġreceive Process Chain LogFile ( channel ) ĊĠĠĠĠĠĠĠ ĠassertThat ( header 1 ). isEqualTo ( json Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" size " Ġto Ġ 1 1 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" end " Ġto Ġ 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" length " Ġto Ġ 1 L ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ ĠassertThat ( received Contents 1 ). isEqualTo ( contents . substring ( 0 , Ġ 1 )) ĊĊĠĠĠĠĠĠĠ Ġ// Ġrequest Ġthe Ġfirst Ġ 5 Ġbytes ĊĠĠĠĠĠĠĠ Ġval Ġmsg 2 Ġ= Ġjson Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" action " Ġto Ġ" fetch ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" id " Ġto Ġid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" reply Address " Ġto Ġreply Address , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" end " Ġto Ġ 4 ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġvert x . event Bus (). send ( address , Ġmsg 2 ) ĊĊĠĠĠĠĠĠĠ Ġval Ġ( received Contents 2 , Ġheader 2 ) Ġ= Ġreceive Process Chain LogFile ( channel ) ĊĠĠĠĠĠĠĠ ĠassertThat ( header 2 ). isEqualTo ( json Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" size " Ġto Ġ 1 1 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" end " Ġto Ġ 4 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" length " Ġto Ġ 5 L ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ ĠassertThat ( received Contents 2 ). isEqualTo ( contents . substring ( 0 , Ġ 5 )) ĊĊĠĠĠĠĠĠĠ Ġ// Ġrequest Ġ 5 Ġbytes Ġin Ġthe Ġmiddle ĊĠĠĠĠĠĠĠ Ġval Ġmsg 3 Ġ= Ġjson Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" action " Ġto Ġ" fetch ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" id " Ġto Ġid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" reply Address " Ġto Ġreply Address , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 2 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" end " Ġto Ġ 6 ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġvert x . event Bus (). send ( address , Ġmsg 3 ) ĊĊĠĠĠĠĠĠĠ Ġval Ġ( received Contents 3 , Ġheader 3 ) Ġ= Ġreceive Process Chain LogFile ( channel ) ĊĠĠĠĠĠĠĠ ĠassertThat ( header 3 ). isEqualTo ( json Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" size " Ġto Ġ 1 1 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 2 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" end " Ġto Ġ 6 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" length " Ġto Ġ 5 L ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ ĠassertThat ( received Contents 3 ). isEqualTo ( contents . substring ( 2 , Ġ 7 )) ĊĊĠĠĠĠĠĠĠ Ġ// Ġrequest Ġthe Ġlast Ġ 5 Ġbytes ĊĠĠĠĠĠĠĠ Ġval Ġmsg 4 Ġ= Ġjson Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" action " Ġto Ġ" fetch ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" id " Ġto Ġid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" reply Address " Ġto Ġreply Address , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 6 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" end " Ġto Ġ 1 0 ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġvert x . event Bus (). send ( address , Ġmsg 4 ) ĊĊĠĠĠĠĠĠĠ Ġval Ġ( received Contents 4 , Ġheader 4 ) Ġ= Ġreceive Process Chain LogFile ( channel ) ĊĠĠĠĠĠĠĠ ĠassertThat ( header 4 ). isEqualTo ( json Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" size " Ġto Ġ 1 1 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 6 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" end " Ġto Ġ 1 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" length " Ġto Ġ 5 L ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ ĠassertThat ( received Contents 4 ). isEqualTo ( contents . substring ( 6 )) ĊĊĠĠĠĠĠĠĠ Ġ// Ġrequest Ġthe Ġlast Ġ 5 Ġbytes Ġ( end Ġposition Ġ 1 Ġtoo Ġhigh ) ĊĠĠĠĠĠĠĠ Ġval Ġmsg 5 Ġ= Ġjson Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" action " Ġto Ġ" fetch ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" id " Ġto Ġid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" reply Address " Ġto Ġreply Address , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 6 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" end " Ġto Ġ 1 1 ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġvert x . event Bus (). send ( address , Ġmsg 5 ) ĊĊĠĠĠĠĠĠĠ Ġval Ġ( received Contents 5 , Ġheader 5 ) Ġ= Ġreceive Process Chain LogFile ( channel ) ĊĠĠĠĠĠĠĠ ĠassertThat ( header 5 ). isEqualTo ( json Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" size " Ġto Ġ 1 1 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 6 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" end " Ġto Ġ 1 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" length " Ġto Ġ 5 L ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ ĠassertThat ( received Contents 5 ). isEqualTo ( contents . substring ( 6 )) ĊĊĠĠĠĠĠĠĠ Ġ// Ġrequest Ġthe Ġlast Ġ 5 Ġbytes Ġ( end Ġposition Ġmuch Ġtoo Ġhigh ) ĊĠĠĠĠĠĠĠ Ġval Ġmsg 6 Ġ= Ġjson Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" action " Ġto Ġ" fetch ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" id " Ġto Ġid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" reply Address " Ġto Ġreply Address , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 6 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" end " Ġto Ġ 1 0 0 ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġvert x . event Bus (). send ( address , Ġmsg 6 ) ĊĊĠĠĠĠĠĠĠ Ġval Ġ( received Contents 6 , Ġheader 6 ) Ġ= Ġreceive Process Chain LogFile ( channel ) ĊĠĠĠĠĠĠĠ ĠassertThat ( header 6 ). isEqualTo ( json Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" size " Ġto Ġ 1 1 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 6 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" end " Ġto Ġ 1 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" length " Ġto Ġ 5 L ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ ĠassertThat ( received Contents 6 ). isEqualTo ( contents . substring ( 6 )) ĊĊĠĠĠĠĠĠĠ Ġ// Ġrequest Ġthe Ġlast Ġ 5 Ġbytes Ġ( no Ġend Ġposition ) ĊĠĠĠĠĠĠĠ Ġval Ġmsg 7 Ġ= Ġjson Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" action " Ġto Ġ" fetch ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" id " Ġto Ġid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" reply Address " Ġto Ġreply Address , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 6 ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġvert x . event Bus (). send ( address , Ġmsg 7 ) ĊĊĠĠĠĠĠĠĠ Ġval Ġ( received Contents 7 , Ġheader 7 ) Ġ= Ġreceive Process Chain LogFile ( channel ) ĊĠĠĠĠĠĠĠ ĠassertThat ( header 7 ). isEqualTo ( json Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" size " Ġto Ġ 1 1 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 6 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" end " Ġto Ġ 1 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" length " Ġto Ġ 5 L ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ ĠassertThat ( received Contents 7 ). isEqualTo ( contents . substring ( 6 )) ĊĊĠĠĠĠĠĠĠ Ġ// Ġsend Ġinvalid Ġrequest Ġ( end Ġposition Ġless Ġthan Ġstart Ġposition ) ĊĠĠĠĠĠĠĠ Ġval Ġmsg 8 Ġ= Ġjson Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" action " Ġto Ġ" fetch ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" id " Ġto Ġid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" reply Address " Ġto Ġreply Address , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 6 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" end " Ġto Ġ 2 ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġvert x . event Bus (). send ( address , Ġmsg 8 ) ĊĊĠĠĠĠĠĠĠ ĠassertThat Thrown By Ġ{ Ġreceive Process Chain LogFile ( channel ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. matches Ġ{ Ġit Ġis ĠHttpStatus Exception Ġ&& Ġit . statusCode Ġ== Ġ 4 1 6 } ĊĊĠĠĠĠĠĠĠ Ġ// Ġsend Ġinvalid Ġrequest Ġ( start Ġposition Ġout Ġof Ġbounds ) ĊĠĠĠĠĠĠĠ Ġval Ġmsg 9 Ġ= Ġjson Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" action " Ġto Ġ" fetch ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" id " Ġto Ġid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" reply Address " Ġto Ġreply Address , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 1 1 ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġvert x . event Bus (). send ( address , Ġmsg 9 ) ĊĊĠĠĠĠĠĠĠ ĠassertThat Thrown By Ġ{ Ġreceive Process Chain LogFile ( channel ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. matches Ġ{ Ġit Ġis ĠHttpStatus Exception Ġ&& Ġit . statusCode Ġ== Ġ 4 1 6 } ĊĊĠĠĠĠĠĠĠ Ġ// Ġrequest Ġthe Ġlast Ġbyte ĊĠĠĠĠĠĠĠ Ġval Ġmsg 1 0 Ġ= Ġjson Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" action " Ġto Ġ" fetch ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" id " Ġto Ġid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" reply Address " Ġto Ġreply Address , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 1 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" end " Ġto Ġ 1 0 ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġvert x . event Bus (). send ( address , Ġmsg 1 0 ) ĊĊĠĠĠĠĠĠĠ Ġval Ġ( received Contents 1 0 , Ġheader 1 0 ) Ġ= Ġreceive Process Chain LogFile ( channel ) ĊĠĠĠĠĠĠĠ ĠassertThat ( header 1 0 ). isEqualTo ( json Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" size " Ġto Ġ 1 1 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 1 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" end " Ġto Ġ 1 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" length " Ġto Ġ 1 L ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ ĠassertThat ( received Contents 1 0 ). isEqualTo ( contents . substring ( contents . length Ġ- Ġ 1 )) ĊĊĠĠĠĠĠĠĠ Ġ// Ġrequest Ġthe Ġlast Ġbyte Ġ( suffix Ġrequest ) ĊĠĠĠĠĠĠĠ Ġval Ġmsg 1 1 Ġ= Ġjson Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" action " Ġto Ġ" fetch ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" id " Ġto Ġid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" reply Address " Ġto Ġreply Address , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ- 1 ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġvert x . event Bus (). send ( address , Ġmsg 1 1 ) ĊĊĠĠĠĠĠĠĠ Ġval Ġ( received Contents 1 1 , Ġheader 1 1 ) Ġ= Ġreceive Process Chain LogFile ( channel ) ĊĠĠĠĠĠĠĠ ĠassertThat ( header 1 1 ). isEqualTo ( json Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" size " Ġto Ġ 1 1 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 1 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" end " Ġto Ġ 1 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" length " Ġto Ġ 1 L ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ ĠassertThat ( received Contents 1 1 ). isEqualTo ( contents . substring ( contents . length Ġ- Ġ 1 )) ĊĊĠĠĠĠĠĠĠ Ġ// Ġrequest Ġthe Ġlast Ġ 5 Ġbytes Ġ( suffix Ġrequest ) ĊĠĠĠĠĠĠĠ Ġval Ġmsg 1 2 Ġ= Ġjson Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" action " Ġto Ġ" fetch ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" id " Ġto Ġid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" reply Address " Ġto Ġreply Address , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ- 5 ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġvert x . event Bus (). send ( address , Ġmsg 1 2 ) ĊĊĠĠĠĠĠĠĠ Ġval Ġ( received Contents 1 2 , Ġheader 1 2 ) Ġ= Ġreceive Process Chain LogFile ( channel ) ĊĠĠĠĠĠĠĠ ĠassertThat ( header 1 2 ). isEqualTo ( json Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" size " Ġto Ġ 1 1 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 6 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" end " Ġto Ġ 1 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" length " Ġto Ġ 5 L ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ ĠassertThat ( received Contents 1 2 ). isEqualTo ( contents . substring ( contents . length Ġ- Ġ 5 )) ĊĊĠĠĠĠĠĠĠ Ġ// Ġrequest Ġtoo Ġmuch Ġ( suffix Ġrequest ) ĊĠĠĠĠĠĠĠ Ġval Ġmsg 1 3 Ġ= Ġjson Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" action " Ġto Ġ" fetch ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" id " Ġto Ġid , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" reply Address " Ġto Ġreply Address , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ- 2 0 ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġvert x . event Bus (). send ( address , Ġmsg 1 3 ) ĊĊĠĠĠĠĠĠĠ Ġval Ġ( received Contents 1 3 , Ġheader 1 3 ) Ġ= Ġreceive Process Chain LogFile ( channel ) ĊĠĠĠĠĠĠĠ ĠassertThat ( header 1 3 ). isEqualTo ( json Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġobj ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" size " Ġto Ġ 1 1 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" start " Ġto Ġ 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" end " Ġto Ġ 1 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" length " Ġto Ġ 1 1 L ĊĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ ĠassertThat ( received Contents 1 3 ). isEqualTo ( contents ) ĊĊĠĠĠĠĠĠĠ Ġctx . complete Now () ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} Ċ } Ċ