package Ġde . f aw e izz . p oc . test Ċ Ċ import Ġde . f aw e izz . p oc . util .* Ċ import Ġio . ktor . client . features .* Ċ import Ġio . ktor . client . request .* Ċ import Ġio . ktor . client . statement .* Ċ import Ġio . ktor . http .* Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . run Blocking Ċ import Ġorg . apache . avro . Schema Ċ import Ġorg . apache . avro . Schema Builder Ċ import Ġorg . apache . avro . generic . Generic Record Ċ import Ġorg . apache . avro . generic . Generic Record Builder Ċ import Ġorg . apache . kafka . clients . producer . Producer Record Ċ import Ġorg . junit . jupiter . api . BeforeEach Ċ import Ġorg . junit . jupiter . api . Test Ċ import Ġorg . junit . jupiter . api . assert Throws Ċ import Ġjava . time . Duration Ċ import Ġjava . util .* Ċ import Ġkotlin . test . assertEquals Ċ Ċ class ĠShared Topic Creation Test Ġ{ ĊĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġtest Message : ĠGeneric Record ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġschema : ĠSchema ĊĊĠĠĠ Ġ@ BeforeEach ĊĠĠĠ Ġinternal Ġfun ĠsetUp () Ġ{ ĊĠĠĠĠĠĠĠ Ġschema Ġ= ĠSchema Builder . builder () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. record (" test _ schema _ 2 ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. fields () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. name (" uid "). type (). string Type (). no Default () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. name (" some Number "). type (). int Type (). no Default () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. end Record () ĊĊĠĠĠĠĠĠĠ Ġtest Message Ġ= ĠGeneric Record Builder ( schema ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. set (" uid ", ĠUUID . randomUUID (). toString ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. set (" some Number ", Ġ 1 9 9 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` actor Ġshould Ġcreate Ġsource Ġresource `() Ġ{ ĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval ĠaccessToken Ġ= Ġget Jwt ( FIRST _ USER _ NAME , ĠFIRST _ USER _ PASSWORD , ĠKEY CLO AK _ ADDRESS ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġcreate HttpClient (). use Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresponse Ġ= Ġit . post < HttpResponse >(" http :// localhost : 8 0 8 1 / resource ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader (" Authorization ", Ġ" Bearer Ġ${ accessToken . accessToken }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader (" Content - Type ", Ġ" application / json ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbody Ġ= ĠTopic Creation Request ( AL ICE _ SOURCE _ RESOURCE , Ġschema . toString (), Ġ" description ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( HttpStatusCode . OK , Ġresponse . status ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` actor Ġshould Ġbe Ġable Ġto Ġcreate Ġjoined Ġresource `() Ġ{ ĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval ĠaccessToken Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġget Jwt ( FIRST _ USER _ NAME , ĠFIRST _ USER _ PASSWORD , ĠKEY CLO AK _ ADDRESS ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġcreate HttpClient (). use Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcreated Resource Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . post < Joined Resource >(" http :// localhost : 8 0 8 1 / joined - resource ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader (" Authorization ", Ġ" Bearer Ġ${ accessToken . accessToken }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader (" Content - Type ", Ġ" application / json ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbody Ġ= ĠJo ined Resource Creation Request ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġname Ġ= ĠSHARE D _ TOPIC _ NAME , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsource Resource Ġ= Ġ" alice .$ AL ICE _ SOURCE _ RESOURCE ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdescription Ġ= Ġ" a Ġcool Ġresource " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals (" joined _$ SHARED _ TOPIC _ NAME ", Ġcreated Resource . resourceId ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( listOf (" alice "), Ġcreated Resource . own ing Actor Ids ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( emptyList (), Ġcreated Resource . code Execution s ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( schema . toString (), Ġcreated Resource . schema ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` actor Ġshould Ġnot Ġbe Ġable Ġto Ġcreate Ġclient Ġfor Ġjoined Ġtopic `() Ġ{ ĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval ĠaccessToken Ġ= Ġget Jwt ( FIRST _ USER _ NAME , ĠFIRST _ USER _ PASSWORD , ĠKEY CLO AK _ ADDRESS ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfull Topic Name Ġ= Ġ" joined _$ SHARED _ TOPIC _ NAME " ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġcreate HttpClient (). use Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertThrows < Server Response Exception > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . post < HttpResponse >(" http :// localhost : 8 0 8 1 / clients ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader (" Authorization ", Ġ" Bearer Ġ${ accessToken . accessToken }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader (" Content - Type ", Ġ" application / json ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbody Ġ= ĠClient Creation Request ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġname Ġ= Ġ" invalid - client - name ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġinput Resource Ids Ġ= ĠlistOf ( full Topic Name ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoutput Resource Ids Ġ= Ġempty List () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` second Ġactor Ġshould Ġbe Ġable Ġto Ġcreate Ġsource Ġresource `() Ġ{ ĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval ĠaccessToken Ġ= Ġget Jwt ( SECOND _ USER , ĠSE COND _ USER _ PASSWORD , ĠKEY CLO AK _ ADDRESS ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġcreate HttpClient (). use Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresponse Ġ= Ġit . post < HttpResponse >(" http :// localhost : 8 0 8 1 / resource ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader (" Authorization ", Ġ" Bearer Ġ${ accessToken . accessToken }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader (" Content - Type ", Ġ" application / json ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbody Ġ= ĠTopic Creation Request ( BO B _ SOURCE _ RESOURCE , Ġschema . toString (), Ġ" description ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( HttpStatusCode . OK , Ġresponse . status ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` second Ġactor Ġshould Ġbe Ġable Ġto Ġcreate Ġjoined Ġresource Ġjoin Ġrequest Ġand Ġfirst Ġuser Ġshould Ġbe Ġable Ġto Ġaccept Ġit `() Ġ{ ĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval ĠaccessToken Ġ= Ġget Jwt ( SECOND _ USER , ĠSE COND _ USER _ PASSWORD , ĠKEY CLO AK _ ADDRESS ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġcreate HttpClient (). use Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsource Topic Name Ġ= Ġ" bob .$ BO B _ SOURCE _ RESOURCE " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġjoined Topic Name Ġ= Ġ" joined _$ SHARED _ TOPIC _ NAME " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresponse Ġ= Ġit . post < Joined Resource Join Request >(" http :// localhost : 8 0 8 1 / joined - resource - poll ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader (" Authorization ", Ġ" Bearer Ġ${ accessToken . accessToken }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader (" Content - Type ", Ġ" application / json ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbody Ġ= ĠJo ined Topic Join Request Request ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġjoined Topic Name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsource Topic Name ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( source Topic Name , Ġresponse . s ourced Resource ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( joined Topic Name , Ġresponse . target Joined Resource ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( SECOND _ USER , Ġresponse . request ing Actor ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( set Of ( FIRST _ USER _ NAME ), Ġresponse . actors To Accept ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( set Of (), Ġresponse . accepted Act ors ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( set Of (), Ġresponse . decl ined Act ors ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġalice AccessToken Ġ= Ġget Jwt ( FIRST _ USER _ NAME , ĠFIRST _ USER _ PASSWORD , ĠKEY CLO AK _ ADDRESS ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġaccept Response Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . post < Joined Resource Join Request >(" http :// localhost : 8 0 8 1 / joined - resource - poll /${ response . id }/ accept ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader (" Authorization ", Ġ" Bearer Ġ${ alice AccessToken . accessToken }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( source Topic Name , Ġaccept Response . s ourced Resource ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( joined Topic Name , Ġaccept Response . target Joined Resource ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( SECOND _ USER , Ġaccept Response . request ing Actor ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( set Of ( FIRST _ USER _ NAME ), Ġaccept Response . actors To Accept ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( set Of ( FIRST _ USER _ NAME ), Ġaccept Response . accepted Act ors ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( set Of (), Ġaccept Response . decl ined Act ors ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` first Ġactor Ġshould Ġbe Ġable Ġto Ġcreate Ġcode Ġexecution Ġrequest Ġand Ġsecond Ġactor Ġshould Ġbe Ġable Ġto Ġaccept Ġit `() Ġ{ ĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval ĠaccessToken Ġ= Ġget Jwt ( FIRST _ USER _ NAME , ĠFIRST _ USER _ PASSWORD , ĠKEY CLO AK _ ADDRESS ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġcreate HttpClient (). use Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġjoined Topic Name Ġ= Ġ" joined _$ SHARED _ TOPIC _ NAME " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresponse Ġ= Ġit . post < Code Execution Poll >(" http :// localhost : 8 0 8 1 / joined - resource - code - poll ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader (" Authorization ", Ġ" Bearer Ġ${ accessToken . accessToken }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader (" Content - Type ", Ġ" application / json ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbody Ġ= ĠCode Execution Request ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresource Name Ġ= Ġjoined Topic Name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġgit Url Ġ= ĠGIT _ URL , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommit Hash Ġ= ĠCOMMIT _ HASH , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġname Ġ= ĠNAME , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcomment Ġ= ĠCOMMENT ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( joined Topic Name , Ġresponse . resourceName ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( GIT _ URL , Ġresponse . git Url ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( FIRST _ USER _ NAME , Ġresponse . requester ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( COMMIT _ HASH , Ġresponse . commit Hash ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( COMMENT , Ġresponse . comment ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( joined Topic Name , Ġresponse . resourceName ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( set Of ( SECOND _ USER ), Ġresponse . actors To Accept ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( set Of (), Ġresponse . accepted Act ors ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( set Of (), Ġresponse . decl ined Act ors ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsecond User AccessToken Ġ= Ġget Jwt ( SECOND _ USER , ĠSE COND _ USER _ PASSWORD , ĠKEY CLO AK _ ADDRESS ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġaccept Response Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . post < Code Execution Poll >(" http :// localhost : 8 0 8 1 / joined - resource - code - poll /${ response . id }/ accept ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader (" Authorization ", Ġ" Bearer Ġ${ second User AccessToken . accessToken }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( joined Topic Name , Ġaccept Response . resourceName ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( GIT _ URL , Ġaccept Response . git Url ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( FIRST _ USER _ NAME , Ġaccept Response . requester ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( COMMIT _ HASH , Ġaccept Response . commit Hash ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( COMMENT , Ġaccept Response . comment ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( joined Topic Name , Ġaccept Response . resourceName ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( set Of ( SECOND _ USER ), Ġaccept Response . actors To Accept ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( set Of ( SECOND _ USER ), Ġaccept Response . accepted Act ors ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( set Of (), Ġaccept Response . decl ined Act ors ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` custom Ġcode Ġexecution Ġshould Ġhave Ġreplic ated Ġmessages Ġto Ġoutput Ġtopic `() Ġ{ ĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcreate HttpClient (). use Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval ĠaccessToken Ġ= Ġget Jwt ( FIRST _ USER _ NAME , ĠFIRST _ USER _ PASSWORD , Ġ" http :// keycloak : 8 0 8 0 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfull Topic Name Ġ= Ġ" joined _${ SHARED _ TOPIC _ NAME }_${ NAME }_ $ COMMIT _ HASH " ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġclient Info Ġ= Ġit . post < Client Info >(" http :// localhost : 8 0 8 1 / clients ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader (" Authorization ", Ġ" Bearer Ġ${ accessToken . accessToken }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader (" Content - Type ", Ġ" application / json ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbody Ġ= ĠClient Creation Request ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġname Ġ= Ġ" listener - client ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġinput Resource Ids Ġ= ĠlistOf ( full Topic Name ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoutput Resource Ids Ġ= ĠlistOf () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġconsumer Ġ= Ġcreate Test Kafka Consumer ( client Info . consumer Group , Ġclient Info . name , Ġclient Info . secret ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . subscribe ( listOf ( full Topic Name )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . poll ( Duration . of Millis ( 1 0 0 0 )) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfirst User Source Topic Ġ= Ġ"$ FIRST _ USER _ NAME .$ AL ICE _ SOURCE _ RESOURCE " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġemit To Source Topic ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠFIRST _ USER _ NAME , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠFIRST _ USER _ PASSWORD , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" first - user - data ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfirst User Source Topic , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtest Message ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsecond User Source Topic Ġ= Ġ"$ SECOND _ USER .$ BO B _ SOURCE _ RESOURCE " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġemit To Source Topic ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠSE COND _ USER , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠSE COND _ USER _ PASSWORD , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" second - user - data ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsecond User Source Topic , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtest Message ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 1 0 0 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmessages Ġ= Ġconsumer . poll ( Duration . of Millis ( 2 0 0 0 0 )) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmessages . records ( full Topic Name ). forEach Ġ{ Ġmessage Ġ-> Ġprintln ( message . value ()) Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 2 , Ġmessages . count ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( test Message , Ġmessages . records ( full Topic Name ). first (). value (). deserialize ( schema )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( test Message , Ġmessages . records ( full Topic Name ). last (). value (). deserialize ( schema )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġemit To Source Topic ( ĊĠĠĠĠĠĠĠ ĠuserName : ĠString , ĊĠĠĠĠĠĠĠ Ġpassword : ĠString , ĊĠĠĠĠĠĠĠ Ġclient Name : ĠString , ĊĠĠĠĠĠĠĠ Ġtopic Name : ĠString , ĊĠĠĠĠĠĠĠ Ġmessage : ĠGeneric Record ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval ĠaccessToken Ġ= Ġget Jwt ( userName , Ġpassword , Ġ" http :// keycloak : 8 0 8 0 ") ĊĊĠĠĠĠĠĠĠ Ġcreate HttpClient (). use Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġclient Info Ġ= Ġit . post < Client Info >(" http :// localhost : 8 0 8 1 / clients ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader (" Authorization ", Ġ" Bearer Ġ${ accessToken . accessToken }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader (" Content - Type ", Ġ" application / json ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbody Ġ= ĠClient Creation Request ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġname Ġ= Ġclient Name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġinput Resource Ids Ġ= Ġempty List (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoutput Resource Ids Ġ= ĠlistOf ( topic Name ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġproducer Ġ= Ġcreate Test Kafka Producer ( client Info . consumer Group , Ġclient Info . name , Ġclient Info . secret ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfuture Ġ= Ġproducer . send ( Producer Record ( topic Name , Ġmessage . serialize ( schema ))) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ(! future . is Done ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 1 0 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠA LIC E _ SOURCE _ RESOURCE Ġ= Ġ" alice - source - topic 2 1 " ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠB OB _ SOURCE _ RESOURCE Ġ= Ġ" bob - source - topic 2 1 " ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠSHARE D _ TOPIC _ NAME Ġ= Ġ" shared - topic 3 1 2 1 " ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠFIRST _ USER _ NAME Ġ= Ġ" alice " ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠFIRST _ USER _ PASSWORD Ġ= Ġ" alice - secret " ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠSE COND _ USER Ġ= Ġ" bob " ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠSE COND _ USER _ PASSWORD Ġ= Ġ" bob - secret " ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠKEY CLO AK _ ADDRESS Ġ= Ġ" http :// keycloak : 8 0 8 0 " ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠGIT _ URL Ġ= Ġ" https :// github . com / f aw e izz / p oc - joined - resource . git " ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠCOMMIT _ HASH Ġ= Ġ" a 5 5 0 c 6 1 3 b 2 2 a 5 8 2 0 dc 1 b 3 7 7 9 cf 7 a 2 d 4 2 fbd 4 9 7 8 a " ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠCOMMENT Ġ= Ġ" This Ġis Ġa Ġcomment " ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠNAME Ġ= Ġ" Custom - Code - Execution - 1 3 2 " ĊĠĠĠ Ġ} Ċ }