package Ġstorage . mongo Ċ Ċ import Ġcom . mongodb . client . model . Index Options Ċ import Ġcom . mongodb . client . model . Indexes Ċ import Ġcom . mongodb . client . model . Insert Many Options Ċ import Ġconfig . Configuration Reader Ċ import Ġconfig . Job Duplicate Entry Policy Ċ import Ġconfig . On Duplicate Entry Ċ import Ġkotlinx . coroutines . run Blocking Ċ import Ġmu . Kotlin Logging Ċ import Ġobservable . Observable Event Ċ import Ġobservable . Observable Event Notifier Ċ import Ġorg . bson . B son Array Ċ import Ġorg . bson . B son Document Ċ import Ġorg . bson . B son String Ċ import Ġorg . kode in . di . DI Ċ import Ġorg . kode in . di . instance Ċ import Ġorg . lit ote . km ongo . upsert Ċ import Ġprocessing . Processed G am ay un Result Ċ import Ġstorage . Data Repository Ċ Ċ private Ġval Ġlogger Ġ= ĠKotlin Logging . logger Ġ{} Ċ Ċ class ĠMongo Data Repository ( kode in : ĠDI ) Ġ: ĠData Repository Ġ{ ĊĠĠĠ Ġprivate Ġval Ġconfiguration Reader : ĠConfiguration Reader Ġby Ġk ode in . instance () ĊĠĠĠ Ġprivate Ġval Ġmongo Db Settings : ĠMongo Db Settings Ġby Ġk ode in . instance () ĊĠĠĠ Ġprivate Ġval Ġobservable Event Notifier : ĠObservable Event Notifier Ġby Ġk ode in . instance () ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcreate Unique Indexes () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġobservable Event Notifier . subscribe To Event ( Observable Event . CONFIGURATION _ RE START ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcreate Unique Indexes () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġcreate Unique Indexes () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġjobs Ġ= Ġconfiguration Reader . read Jobs Configuration () ĊĠĠĠĠĠĠĠ Ġjobs . forEach Ġ{ Ġjob Config Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( job Config . job Duplicate Entry Policy ?. unique Ids Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Creating Ġunique Ġindex Ġfor Ġcollection Ġ${ job Config . name }. ĠInd ices Ġare : Ġ${ job Config . job Duplicate Entry Policy . unique Ids }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcollection Ġ= Ġmongo Db Settings . database . get Collection < B son Document >( job Config . name ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcollection . create Index ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠIndexes . ascending ( job Config . job Duplicate Entry Policy . unique Ids ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠIndex Options (). unique ( true ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġstore Result ( ĊĠĠĠĠĠĠĠ Ġjob Id : ĠString , ĊĠĠĠĠĠĠĠ Ġresult : ĠList < Processed G am ay un Result >, ĊĠĠĠĠĠĠĠ Ġduplicate Entry Policy : ĠJob Duplicate Entry Policy ? ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( duplicate Entry Policy Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstore With No Duplicate Policy ( jobId , Ġresult ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( duplicate Entry Policy . on Duplicate Entry Policy ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠOn Duplicate Entry . IGNORE _ NEW Ġ-> Ġstore With Duplicate Policy Ignore New ( jobId , Ġresult ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠOn Duplicate Entry . STORE _ NEW Ġ-> Ġstore With Duplicate Policy Store New ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġjob Id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresult , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġduplicate Entry Policy . unique Ids ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠOn Duplicate Entry . TRACK _ CHANG ES Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġstore With Duplicate Policy Ignore New ( jobId : ĠString , Ġresult : ĠList < Processed G am ay un Result >) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcollection Ġ= Ġmongo Db Settings . database . get Collection < B son Document >( jobId ) ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// if Ġordered Ġis Ġset Ġto Ġtrue Ġ( default Ġvalue ) Ġentire Ġwrite Ġfails Ġif Ġwrite Ġfor Ġa Ġsingle Ġdocument Ġfails ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// failed Ġwrites Ġfor Ġsingle Ġdocuments Ġcan Ġhappen Ġhere Ġbecause Ġwe Ġmight Ġbe Ġtrying Ġto Ġwrite Ġa Ġsame Ġdocument Ġmultiple Ġtimes ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ( for Ġexample , Ġif Ġwe Ġparse Ġthe Ġsame Ġad Ġin Ġtwo Ġdifferent Ġscheduler Ġruns ) Ġif Ġunique Index Ġis Ġused ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// there fore , Ġwe Ġdon 't Ġwant Ġeverything Ġto Ġfail Ġif Ġsome Ġdocuments Ġfail , Ġso Ġwe Ġare Ġsetting Ġordered Ġto Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// see : Ġhttps :// docs . mongodb . com / php - library / master / reference / method / MongoDB Collection - insert Many / ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcollection . insert Many ( result . map Ġ{ Ġit . to B son Document () Ġ}, ĠInsert Many Options (). ordered ( false )) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : Ġcom . mongodb . Mongo Bulk Write Exception ) Ġ{ Ġ// todo : Ġthink Ġabout Ġcatch ing Ġsomething Ġmore Ġspecific Ġhere Ġor Ġlogging Ġwarning Ġif Ġit Ġis Ġnot Ġrewrite Ġerror ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// this Ġexception Ġhappens Ġwhen Ġwe Ġare Ġtrying Ġto Ġinsert Ġa Ġdocument Ġwhose Ġunique ĠID Ġis Ġalready Ġpresent Ġin ĠMongoDB ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// this Ġactually Ġhappens Ġoften Ġin ĠG am ay un Ġscenario Ġand Ġit Ġis Ġnot Ġan Ġerror ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug ( e . toString ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġstore With Duplicate Policy Store New ( ĊĠĠĠĠĠĠĠ Ġjob Id : ĠString , ĊĠĠĠĠĠĠĠ Ġresult : ĠList < Processed G am ay un Result >, ĊĠĠĠĠĠĠĠ Ġunique Ids : ĠList < String > ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcollection Ġ= Ġmongo Db Settings . database . get Collection < B son Document >( jobId ) ĊĠĠĠĠĠĠĠ Ġresult . forEach Ġ{ Ġdocument Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġkeys With Values Ġ= Ġunique Ids ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġunique Id Ġ-> ĠPair ( unique Id , Ġdocument . string Data [ unique Id ]) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġit . second Ġ!= Ġnull Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. toMap () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġupdate Keys Matcher Ġ= ĠB son Document (). also Ġ{ Ġdocument Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġkeys With Values . forEach Ġ{ Ġkey WithValue Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdocument [ key WithValue . key ] Ġ= ĠB son String ( key WithValue . value ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}. toJson () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġcollection . update One ( update Keys Matcher , Ġdocument , Ġupsert ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ// for Ġmongo Ġwe Ġcan Ġ Ġtreat Ġthese Ġtwo Ġsituations Ġas Ġthe Ġsame ĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġstore With No Duplicate Policy ( jobId : ĠString , Ġresult : ĠList < Processed G am ay un Result >) Ġ= ĊĠĠĠĠĠĠĠ Ġstore With Duplicate Policy Ignore New ( jobId , Ġresult ) ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġstore With Duplicate Policy Track Changes ( ĊĠĠĠĠĠĠĠ Ġjob Id : ĠString , ĊĠĠĠĠĠĠĠ Ġresult : ĠList < Processed G am ay un Result >, ĊĠĠĠĠĠĠĠ Ġunique Ids : ĠList < String > ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ ĠTODO (" This Ġfeature Ġneeds Ġfurther Ġthinking Ġon Ġhow Ġexactly Ġwe Ġwant Ġthe Ġdata Ġstored Ġin Ġthe ĠDB ") ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġprivate Ġfun ĠPro cessed G am ay un Result . to B son Document (): ĠB son Document Ġ= ĊĠĠĠĠĠĠĠ ĠB son Document (). also Ġ{ Ġdocument Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstring List Data . forEach Ġ{ Ġkey ValuePair Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdocument [ key ValuePair . key ] Ġ= ĠB son Array (). also Ġ{ Ġarray Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġkey ValuePair . value . forEach Ġ{ Ġvalue Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġarray . add ( B son String ( value )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġstring Data . forEach Ġ{ Ġkey ValuePair Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdocument [ key ValuePair . key ] Ġ= ĠB son String ( key ValuePair . value ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ċ Ċ }