/* Ċ Ġ* ĠLicensed Ġto Ġthe ĠApache ĠSoftware ĠFoundation Ġ( ASF ) Ġunder Ġone Ċ Ġ* Ġor Ġmore Ġcontributor Ġlicense Ġagreements . Ġ ĠSee Ġthe ĠNOTICE Ġfile Ċ Ġ* Ġdistributed Ġwith Ġthis Ġwork Ġfor Ġadditional Ġinformation Ċ Ġ* Ġregarding Ġcopyright Ġownership . Ġ ĠThe ĠASF Ġlicenses Ġthis Ġfile Ċ Ġ* Ġto Ġyou Ġ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 Ġio . f lood plain . integration Ċ Ċ import Ġcom . mongodb . client . Mongo Clients Ċ import Ġio . f lood plain . kot l ind sl . each Ċ import Ġio . f lood plain . kot l ind sl . from Ċ import Ġio . f lood plain . kot l ind sl . join Ċ import Ġio . f lood plain . kot l ind sl . join Remote Ċ import Ġio . f lood plain . kot l ind sl . message . empty Ċ import Ġio . f lood plain . jdbc . postgres Source Ċ import Ġio . f lood plain . jdbc . postgres Source Config Ċ import Ġio . f lood plain . kot l ind sl . scan Ċ import Ġio . f lood plain . kot l ind sl . set Ċ import Ġio . f lood plain . kot l ind sl . stream Ċ import Ġio . f lood plain . kot l ind sl . to Topic Ċ import Ġio . f lood plain . mongodb . remote Mongo Config Ċ import Ġio . f lood plain . mongodb . to Mongo Ċ import Ġio . f lood plain . mongodb . waitFor Mongo Db Condition Ċ import Ġio . f lood plain . test . Instanti ated Container Ċ import Ġio . f lood plain . test . use Integr aton Ċ import Ġkotlinx . coroutines . cancel Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . with Timeout Ċ import Ġorg . junit . jupiter . api . After All Ċ import Ġorg . junit . jupiter . api . Assertions .* Ċ import Ġorg . junit . jupiter . api . Test Ċ import Ġorg . junit . jupiter . api . Test Instance Ċ import Ġjava . math . BigDecimal Ċ import Ġjava . util . Date Ċ Ċ private Ġval Ġlogger Ġ= Ġmu . Kotlin Logging . logger Ġ{} Ċ Ċ @ kotlin x . coroutines . Experimental Cor outines Api Ċ @ Test Instance ( Test Instance . Lifecycle . PER _ CLASS ) Ċ class ĠPostgres To MongoDB Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġpostgres Container Ġ= ĠInstanti ated Container (" f lood plain / f lood plain - postgres - demo : 1 . 0 . 0 ", Ġ 5 4 3 2 ) ĊĠĠĠ Ġprivate Ġval Ġmongo Container Ġ= ĠInstanti ated Container (" mongo : latest ", Ġ 2 7 0 1 7 ) ĊĊĠĠĠ Ġ@ After All ĊĠĠĠ Ġfun Ġshutdown () Ġ{ ĊĠĠĠĠĠĠĠ Ġpostgres Container . close () ĊĠĠĠĠĠĠĠ Ġmongo Container . close () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠTest Ġthe Ġsimplest Ġimag in able Ġpipe : ĠOne Ġsource Ġand Ġone Ġsink . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Postgres Source Simp ler () Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ(! use Integr aton ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Not Ġperforming Ġintegration Ġtests ; Ġdoesn 't Ġseem Ġto Ġwork Ġin Ġcircle ci ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġstream (" some tenant ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġpostgres Config Ġ= Ġpostgres Source Config ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" my postgres ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpostgres Container . host , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpostgres Container . ex posed Port , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" postgres ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" my se cre tp assword ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" dvd rent al ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" public " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġval Ġlog Config Ġ= Ġlog Sink Config (" any ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmongo Config Ġ= Ġremote Mongo Config ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" mong os ink ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" mongodb ://${ mongo Container . host }:${ mongo Container . ex posed Port }", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" mong od ump " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpostgres Source (" address ", Ġpostgres Config ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġjoin Remote ({ Ġmsg Ġ-> Ġ"${ msg [" city _ id "] }" Ġ}, Ġfalse ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpostgres Source (" city ", Ġpostgres Config ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset Ġ{ Ġ_, Ġmsg , Ġstate Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg [" city "] Ġ= Ġstate ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġto Mongo (" address ", Ġ"$ generation - sink topic address ", Ġmongo Config ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}. render And Execute Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdatabase Ġ= Ġtopology Context (). topic Name (" mong od ump ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflush Sink s () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġhits Ġ= ĠwaitFor Mongo Db Condition ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" mongodb ://${ mongo Container . host }:${ mongo Container . ex posed Port }", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdatabase , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 1 3 0 0 0 0 L ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) Ġ{ Ġdatabase Instance Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcollection Ġ= Ġdatabase Instance . get Collection (" address ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcount Documents Ġ= Ġcollection . count Documents () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Output Ġtopic Ġsize : Ġ${ this . output Size ("$ generation - sink topic address ") }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info ("# Ġof Ġdocuments : Ġ$ count Documents ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( count Documents Ġ== Ġ 6 0 3 L ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 6 0 3 L ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġas ĠLong ? ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertNotNull ( hits ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconnect Jobs (). forEach Ġ{ Ġit . cancel (" c iao !") Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" done , Ġtest Ġsucceeded ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest MongoDB DataType () Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ(! use Integr aton ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Not Ġperforming Ġintegration Ġtests ; Ġdoesn 't Ġseem Ġto Ġwork Ġin Ġcircle ci ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġstream (" some tenant ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġval Ġlog Config Ġ= Ġlog Sink Config (" any ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmongo Config Ġ= Ġremote Mongo Config ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" mong os ink ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" mongodb ://${ mongo Container . host }:${ mongo Container . ex posed Port }", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ"@ mong od ump " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfrom (" input ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġto Mongo (" testdata structure ", Ġ" output ", Ġmongo Config ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}. render And Execute Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdate Input Ġ= ĠDate () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġinput Message Ġ= Ġempty (). set (" date ", Ġdate Input ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġaa Ġ= Ġinput Message . to Immutable () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprintln ("> > Ġ$ aa ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġinput (" input ", Ġ" key 1 ", Ġempty (). set (" date ", Ġdate Input )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdatabase Ġ= Ġtopology Context (). topic Name ("@ mong od ump ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflush Sink s () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġhits Ġ= ĠwaitFor Mongo Db Condition ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" mongodb ://${ mongo Container . host }:${ mongo Container . ex posed Port }", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdatabase ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) Ġ{ Ġdatabase Instance Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcollection Ġ= Ġdatabase Instance . get Collection (" testdata structure ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcount Documents Ġ= Ġcollection . count Documents () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info ("# Ġof Ġdocuments : Ġ$ count Documents ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( count Documents Ġ== Ġ 1 L ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdate Object Ġ= Ġcollection . find (). first ()?. get (" date ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Date : Ġ$ date Object ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( date Input . to Instant (). to Epoch Milli (), Ġdate Object Ġas ĠLong ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 1 L ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġas ĠLong ? ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertNotNull ( hits ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconnect Jobs (). forEach Ġ{ Ġit . cancel (" c iao !") Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" done , Ġtest Ġsucceeded ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠTest Ġthe Ġsimplest Ġimag in able Ġpipe : ĠOne Ġsource Ġand Ġone Ġsink . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Postgres Source () Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ(! use Integr aton ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Not Ġperforming Ġintegration Ġtests ; Ġdoesn 't Ġseem Ġto Ġwork Ġin Ġcircle ci ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġstream Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġpostgres Config Ġ= Ġpostgres Source Config ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" my postgres ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpostgres Container . host , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpostgres Container . ex posed Port , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" postgres ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" my se cre tp assword ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" dvd rent al ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" public " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmongo Config Ġ= Ġremote Mongo Config ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" mong os ink ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" mongodb ://${ mongo Container . host }:${ mongo Container . ex posed Port }", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ"$ generation - mong od ump " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpostgres Source (" address ", Ġpostgres Config ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġjoin Remote ({ Ġmsg Ġ-> Ġ"${ msg [" city _ id "] }" Ġ}, Ġfalse ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpostgres Source (" city ", Ġpostgres Config ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġjoin Remote ({ Ġmsg Ġ-> Ġ"${ msg [" country _ id "] }" Ġ}, Ġfalse ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpostgres Source (" country ", Ġpostgres Config ) Ġ{} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset Ġ{ Ġ_, Ġmsg , Ġstate Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg . set (" country ", Ġstate ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset Ġ{ Ġ_, Ġmsg , Ġstate Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg . set (" city ", Ġstate ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġto Topic ("$ generation - address ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpostgres Source (" customer ", Ġpostgres Config ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġjoin Remote ({ Ġm Ġ-> Ġ"${ m [" address _ id "] }" Ġ}, Ġfalse ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfrom ("$ generation - address ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset Ġ{ Ġ_, Ġmsg , Ġstate Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg . set (" address ", Ġstate ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġto Mongo (" customer ", Ġ"$ generation - customer ", Ġmongo Config ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpostgres Source (" store ", Ġpostgres Config ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġjoin Remote ({ Ġm Ġ-> Ġ"${ m [" address _ id "] }" Ġ}, Ġfalse ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfrom ("$ generation - address ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset Ġ{ Ġ_, Ġmsg , Ġstate Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg . set (" address ", Ġstate ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġto Mongo (" store ", Ġ"$ generation - store ", Ġmongo Config ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpostgres Source (" staff ", Ġpostgres Config ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġjoin Remote ({ Ġm Ġ-> Ġ"${ m [" address _ id "] }" Ġ}, Ġfalse ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfrom ("$ generation - address ") Ġ{} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset Ġ{ Ġ_, Ġmsg , Ġstate Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg [" address "] Ġ= Ġstate ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg [" address _ id "] Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġto Mongo (" staff ", Ġ"$ generation - staff ", Ġmongo Config ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}. render And Execute Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdatabase Ġ= Ġ"${ this . topology Context . generation }- mong od ump " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflush Sink s () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġhits Ġ= ĠwaitFor Mongo Db Condition ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" mongodb ://${ mongo Container . host }:${ mongo Container . ex posed Port }", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdatabase ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) Ġ{ Ġdatabase Instance Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstaff Ġ= Ġdatabase Instance . get Collection (" staff ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstaff Count Ġ= Ġstaff . count Documents () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstore Ġ= Ġdatabase Instance . get Collection (" store ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstore Count Ġ= Ġstore . count Documents () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcustomer Ġ= Ġdatabase Instance . get Collection (" customer ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcustomer Count Ġ= Ġcustomer . count Documents () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info ("# Ġof Ġstaff : Ġ$ staff Count Ġ# Ġof Ġstores : Ġ$ store Count Ġcustomer Ġcount : Ġ$ customer Count ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( store Count Ġ== Ġ 2 L Ġ&& Ġstaff Count Ġ== Ġ 2 L Ġ&& Ġcustomer Count Ġ== Ġ 5 9 9 L ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcustomer Count ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġas ĠLong ? ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertNotNull ( hits ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconnect Jobs (). forEach Ġ{ Ġit . cancel (" c iao !") Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Simple Reduce () Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ(! use Integr aton ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Not Ġperforming Ġintegration Ġtests ; Ġdoesn 't Ġseem Ġto Ġwork Ġin Ġcircle ci ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġstream Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġpostgres Config Ġ= Ġpostgres Source Config ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" my postgres ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpostgres Container . host , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpostgres Container . ex posed Port , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" postgres ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" my se cre tp assword ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" dvd rent al ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" public " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmongo Config Ġ= Ġremote Mongo Config ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" mong os ink ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" mongodb ://${ mongo Container . host }:${ mongo Container . ex posed Port }", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ"$ generation - mong od ump " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpostgres Source (" payment ", Ġpostgres Config ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġscan ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ{ Ġempty (). set (" total ", ĠBigDecimal ( 0 )) Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset Ġ{ Ġ_, Ġmsg , Ġstate Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstate [" total "] Ġ= Ġ( state [" total "] Ġas ĠBigDecimal ). add ( msg [" amount "] Ġas ĠBigDecimal ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstate ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset Ġ{ Ġ_, Ġmsg , Ġstate Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstate [" total "] Ġ= Ġ( state [" total "] Ġas ĠBigDecimal ). subtract ( msg [" amount "] Ġas ĠBigDecimal ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstate ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġto Mongo (" just total ", Ġ"$ generation - my final topic ", Ġmongo Config ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}. render And Execute Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdatabase Ġ= Ġtopology Context (). topic Name ("@ mong od ump ") ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġwith Timeout ( 1 0 0 0 0 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrepeat ( 1 0 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMongo Clients . create (" mongodb ://${ mongo Container . host }:${ mongo Container . ex posed Port }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. use Ġ{ Ġclient Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcollection Ġ= Ġclient . get Database ( database ). get Collection (" just total ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtotal Ġ= Ġcollection . find (). first ()?. get (" total ") Ġas ĠDouble ? Ġ?: Ġ 0 . 0 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( total > Ġ 6 1 3 1 2 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ with Timeout ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Current Ġtotal : Ġ$ total ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 1 0 0 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Test Ġdone , Ġtotal Ġcomputed ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Payment Per Customer () Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ(! use Integr aton ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Not Ġperforming Ġintegration Ġtests ; Ġdoesn 't Ġseem Ġto Ġwork Ġin Ġcircle ci ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġstream Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġpostgres Config Ġ= Ġpostgres Source Config ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" my postgres ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpostgres Container . host , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpostgres Container . ex posed Port , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" postgres ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" my se cre tp assword ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" dvd rent al ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" public " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmongo Config Ġ= Ġremote Mongo Config ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" mong os ink ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" mongodb ://${ mongo Container . host }:${ mongo Container . ex posed Port }", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ"$ generation - mong od ump " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpostgres Source (" customer ", Ġpostgres Config ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġeach Ġ{ Ġ_, Ġm ms , Ġ_ Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Customer : Ġ$ m ms ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġjoin Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpostgres Source (" payment ", Ġpostgres Config ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġscan ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ{ Ġmsg Ġ-> Ġmsg [" customer _ id "]. toString () Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ{ Ġempty (). set (" total ", ĠBigDecimal ( 0 )) Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset Ġ{ Ġ_, Ġmsg , Ġstate Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstate [" total "] Ġ= Ġ( state [" total "] Ġas ĠBigDecimal ). add ( msg [" amount "] Ġas ĠBigDecimal ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstate ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset Ġ{ Ġ_, Ġmsg , Ġstate Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstate [" total "] Ġ= Ġ( state [" total "] Ġas ĠBigDecimal ). add ( msg [" amount "] Ġas ĠBigDecimal ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstate ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset Ġ{ Ġ_, Ġcustomer , Ġpayment total Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcustomer [" payments "] Ġ= Ġpayment total [" total "]; Ġcustomer ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġto Mongo (" payment per customer ", Ġ"$ generation - my final topic ", Ġmongo Config ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}. render And Execute Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġitems Ġ= ĠwaitFor Mongo Db Condition ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" mongodb ://${ mongo Container . host }:${ mongo Container . ex posed Port }", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ"${ topology Context . generation }- mong od ump " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) Ġ{ Ġcurrent Database Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcollection Ġ= Ġcurrent Database . get Collection (" payment per customer ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġitems Ġ= Ġcollection . count Documents () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġitems > Ġ 1 0 0 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertNotNull ( items ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ