package Ġorg . en ode framework . command ing . impl Ċ Ċ import Ġcom . google . common . base . Strings Ċ import Ġkotlinx . coroutines . Coroutine Scope Ċ import Ġkotlinx . coroutines . Dispatch ers Ċ import Ġkotlinx . coroutines . async Ċ import Ġkotlinx . coroutines . future . as Comp letableFuture Ċ import Ġorg . en ode framework . command ing .* Ċ import Ġorg . en ode framework . configurations . Sys Properties Ċ import Ġorg . en ode framework . common . io . IO Helper Ċ import Ġorg . en ode framework . common . io . Task Ċ import Ġorg . en ode framework . common . serializing . I Serialize Service Ċ import Ġorg . en ode framework . domain . Aggregate Root Reference Changed Exception Ċ import Ġorg . en ode framework . domain . I Aggregate Root Ċ import Ġorg . en ode framework . domain . ID omain Exception Ċ import Ġorg . en ode framework . domain . I Memory Cache Ċ import Ġorg . en ode framework . event ing .* Ċ import Ġorg . en ode framework . infrastructure . I TypeName Provider Ċ import Ġorg . en ode framework . messaging . I Application Message Ċ import Ġorg . en ode framework . messaging . I Message Publisher Ċ import Ġorg . en ode framework . messaging . MessageHandler Data Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġjava . util .* Ċ import Ġjava . util . concurrent . Comp letableFuture Ċ import Ġjava . util . concurrent . Completion Exception Ċ import Ġjava . util . function . Function Ċ Ċ /** Ċ Ġ* Ġ@ author Ġan ru ence @ gmail . com Ċ Ġ*/ Ċ class ĠDefault Processing CommandHandler ( private Ġval Ġevent Store : ĠI Event Store , Ġprivate Ġval Ġcommand Handler Provider : ĠICommand Handler Provider , Ġprivate Ġval ĠtypeName Provider : ĠI TypeName Provider , Ġprivate Ġval Ġevent Commit ting Service : ĠI Event Commit ting Service , Ġprivate Ġval Ġmemory Cache : ĠI Memory Cache , Ġprivate Ġval Ġapplication Message Publisher : ĠI Message Publisher < I Application Message >, Ġprivate Ġval Ġexception Publisher : ĠI Message Publisher < ID omain Exception >, Ġprivate Ġval Ġserialize Service : ĠI Serialize Service ) Ġ: ĠI Processing CommandHandler Ġ{ ĊĠĠĠ Ġoverride Ġfun Ġhandle Async ( processing Command : ĠProcessing Command ): ĠCompletableFuture < Boolean > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcommand Ġ= Ġprocessing Command . message ĊĠĠĠĠĠĠĠ Ġif Ġ( Strings . isNull OrEmpty ( command . aggregate Root Id )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval ĠerrorMessage Ġ= ĠString . format (" The Ġaggregate Root Id Ġof Ġcommand Ġcannot Ġbe Ġnull Ġor Ġempty . Ġcommand Type :% s , Ġcommand Id :% s ", Ġcommand . java Class . name , Ġcommand . id ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error ( errorMessage ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġcomplete Command ( processing Command , ĠCommand Status . Failed , ĠString :: class . java . name , ĠerrorMessage ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġfind Result Ġ= Ġget CommandHandler ( processing Command ) Ġ{ Ġcommand Type : ĠClass < *> Ġ-> Ġcommand Handler Provider . get Handlers ( command Type ) Ġ} ĊĠĠĠĠĠĠĠ Ġwhen Ġ( find Result . find Status ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠHandler Find Status . Found Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġhandle Command Internal ( processing Command , Ġfind Result . find Handler Ġas ĠICommand Handler Proxy , Ġ 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠHandler Find Status . Too Many Handler Data Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Found Ġmore Ġthan Ġone Ġcommand Ġhandler Ġdata , Ġcommand Type : {}, Ġcommand Id :{} ", Ġcommand . java Class . name , Ġcommand . id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġcomplete Command ( processing Command , ĠCommand Status . Failed , ĠString :: class . java . name , Ġ" More Ġthan Ġone Ġcommand Ġhandler Ġdata Ġfound .") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠHandler Find Status . Too Many Handler Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Found Ġmore Ġthan Ġone Ġcommand Ġhandler , Ġcommand Type : {}, Ġcommand Id :{} ", Ġcommand . java Class . name , Ġcommand . id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġcomplete Command ( processing Command , ĠCommand Status . Failed , ĠString :: class . java . name , Ġ" More Ġthan Ġone Ġcommand Ġhandler Ġfound .") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠHandler Find Status . NotFound Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval ĠerrorMessage Ġ= ĠString . format (" No Ġcommand Ġhandler Ġfound Ġof Ġcommand . Ġcommand Type :% s , Ġcommand Id :% s ", Ġcommand . java Class . name , Ġcommand . id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error ( errorMessage ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġcomplete Command ( processing Command , ĠCommand Status . Failed , ĠString :: class . java . name , ĠerrorMessage ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġreturn ĠTask . completed Task ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġhandle Command Internal ( processing Command : ĠProcessing Command , Ġcommand Handler : ĠICommand Handler Proxy , Ġretry Times : ĠInt ): ĠCompletableFuture < Boolean > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcommand Ġ= Ġprocessing Command . message ĊĠĠĠĠĠĠĠ Ġval Ġcommand Context Ġ= Ġprocessing Command . command Execute Context ĊĠĠĠĠĠĠĠ Ġif Ġ( processing Command . is D uplicated ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġre publish Command Events ( processing Command , Ġ 0 ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġtask Source Ġ= ĠCompletableFuture < Boolean >() ĊĠĠĠĠĠĠĠ ĠIO Helper . try Async Action Re cursively Without Result (" Handle Command Async ", Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcommand Context . clear () ĊĠĠĠĠĠĠĠĠĠĠĠ ĠCor outine Scope ( Dispatch ers . IO ). async Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommand Handler . handle Async ( command Context , Ġcommand ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}. as Comp letableFuture () ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( logger . isDebugEnabled ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug (" Handle Ġcommand Ġsuccess . Ġhandler Type : {}, Ġcommand Type : {}, Ġcommand Id : {}, Ġaggregate Root Id :{} ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommand Handler . get Inner Object (). java Class . name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommand . java Class . name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommand . id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommand . aggregate Root Id ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( command Context . application Message Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommit Changes Async ( processing Command , Ġtrue , Ġcommand Context . application Message , Ġ"") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. when Complete Ġ{ Ġ_, Ġ_ Ġ-> Ġtask Source . complete ( true ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommit Aggregate Changes ( processing Command ). when Complete Ġ{ Ġ_, Ġ_ Ġ-> Ġtask Source . complete ( true ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. exception ally Ġ{ Ġex : ĠThrowable Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Commit Ġaggregate Ġchanges Ġhas Ġunknown Ġexception , Ġthis Ġshould Ġnot Ġbe Ġhappen , Ġand Ġwe Ġjust Ġcomplete Ġthe Ġcommand , Ġhandler Type : {}, Ġcommand Type : {}, Ġcommand Id : {}, Ġaggregate Root Id :{} ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommand Handler . get Inner Object (). java Class . name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommand . java Class . name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommand . id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommand . aggregate Root Id , Ġex ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcomplete Command ( processing Command , ĠCommand Status . Failed , Ġex . java Class . name , Ġ" Unknown Ġexception Ġcaught Ġwhen Ġcommit ting Ġchanges Ġof Ġcommand .") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. when Complete Ġ{ Ġ_, Ġ_ Ġ-> Ġtask Source . complete ( true ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( aggregate Root Reference Changed Exception : ĠAggregate Root Reference Changed Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Aggregate Ġroot Ġreference Ġchanged Ġwhen Ġprocessing Ġcommand , Ġtry Ġto Ġre - handle Ġthe Ġcommand . Ġaggregate Root Id : Ġ{}, Ġaggregate Root Type : Ġ{}, Ġcommand Id : Ġ{}, Ġcommand Type : Ġ{}, Ġhandler Type : Ġ{}", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaggregate Root Reference Changed Exception . aggregate Root . unique Id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaggregate Root Reference Changed Exception . aggregate Root . java Class . name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommand . id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommand . java Class . name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommand Handler . get Inner Object (). java Class . name ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġhandle Command Internal ( processing Command , Ġcommand Handler , Ġ 0 ). when Complete Ġ{ Ġ_, Ġ_ Ġ-> Ġtask Source . complete ( true ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Commit Ġaggregate Ġchanges Ġhas Ġunknown Ġexception , Ġthis Ġshould Ġnot Ġbe Ġhappen , Ġand Ġwe Ġjust Ġcomplete Ġthe Ġcommand , Ġhandler Type : {}, Ġcommand Type : {}, Ġcommand Id : {}, Ġaggregate Root Id :{} ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommand Handler . get Inner Object (). java Class . name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommand . java Class . name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommand . id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommand . aggregate Root Id , Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcomplete Command ( processing Command , ĠCommand Status . Failed , Ġe . java Class . name , Ġ" Unknown Ġexception Ġcaught Ġwhen Ġcommit ting Ġchanges Ġof Ġcommand .") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. when Complete Ġ{ Ġ_, Ġ_ Ġ-> Ġtask Source . complete ( true ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠString . format ("[ command :[ id :% s , type :% s ], handler Type :% s , aggregate Root Id :% s ]", Ġcommand . id , Ġcommand . java Class . name , Ġcommand Handler . get Inner Object (). java Class . name , Ġcommand . aggregate Root Id ) ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ Ġex : ĠThrowable , ĠerrorMessage : ĠString Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġhandle Exception Async ( processing Command , Ġcommand Handler , Ġex , ĠerrorMessage , Ġ 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. when Complete Ġ{ Ġ_, Ġ_ Ġ-> Ġtask Source . complete ( true ) Ġ} ĊĠĠĠĠĠĠĠ Ġ}, Ġretry Times , Ġfalse ) ĊĠĠĠĠĠĠĠ Ġreturn Ġtask Source ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcommit Aggregate Changes ( processing Command : ĠProcessing Command ): ĠCompletableFuture < Boolean > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcommand Ġ= Ġprocessing Command . message ĊĠĠĠĠĠĠĠ Ġval Ġcontext Ġ= Ġprocessing Command . command Execute Context ĊĠĠĠĠĠĠĠ Ġval Ġtracked Aggregate Roots Ġ= Ġcontext . tracked Aggregate Roots ĊĠĠĠĠĠĠĠ Ġvar Ġdirty Aggregate Root Count Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġvar Ġdirty Aggregate Root : ĠI Aggregate Root ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġvar Ġchanged Events : ĠList < ID omain Event <* >> Ġ= ĠArrayList () ĊĠĠĠĠĠĠĠ Ġfor Ġ( aggregate Root Ġin Ġtracked Aggregate Roots ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġevents Ġ= Ġaggregate Root . changes ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( events . size Ġ> Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdirty Aggregate Root Count ++ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( dirty Aggregate Root Count Ġ> Ġ 1 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval ĠerrorMessage Ġ= ĠString . format (" Detected Ġmore Ġthan Ġone Ġaggregate Ġcreated Ġor Ġmodified Ġby Ġcommand . Ġcommand Type :% s , Ġcommand Id :% s ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommand . java Class . name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommand . id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error ( errorMessage ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġcomplete Command ( processing Command , ĠCommand Status . Failed , ĠString :: class . java . name , ĠerrorMessage ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdirty Aggregate Root Ġ= Ġaggregate Root ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġchanged Events Ġ= Ġevents ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ// å¦Ĥæŀľ å½ĵåīį command æ²¡æľī å¯¹ ä»»ä½ķ èģļ åĲĪ æł¹ åģļ ä¿®æĶ¹ ï¼Į æ¡Ĩæŀ¶ ä»į çĦ¶ éľĢè¦ģ å°Ŀè¯ķ èİ·åıĸ è¯¥ command ä¹ĭåīį æĺ¯åĲ¦ æľī äº§çĶŁ äºĭä»¶ ï¼Į ĊĠĠĠĠĠĠĠ Ġ// å¦Ĥæŀľ æľī ï¼Į åĪĻ éľĢè¦ģ å°Ĩ äºĭä»¶ åĨį æ¬¡ åıĳå¸ĥ åĪ° MQ ï¼Ľ å¦Ĥæŀľ æ²¡æľī ï¼Į åĪĻ å®ĮæĪĲ åĳ½ä»¤ ï¼Į è¿ĶåĽŀ command çļĦ ç»ĵæŀľ ä¸º Nothing Changed ãĢĤ ĊĠĠĠĠĠĠĠ Ġ// ä¹ĭ æīĢä»¥ è¦ģ è¿Ļæł· åģļ æĺ¯ åĽłä¸º æľī åı¯èĥ½ å½ĵåīį command ä¸Ĭ æ¬¡ æī§è¡Į çļĦ ç»ĵæŀľ åı¯èĥ½ æĺ¯ äºĭä»¶ æĮģ ä¹ħ åĮĸ å®ĮæĪĲ ï¼Į ä½Ĩæĺ¯ åıĳå¸ĥ åĪ° MQ æľª å®ĮæĪĲ ï¼Į çĦ¶åĲİ éĤ£ æĹ¶ æŃ£ å¥½ æľºåĻ¨ æĸŃ çĶµ å® ķ æľº äºĨ ï¼Ľ ĊĠĠĠĠĠĠĠ Ġ// è¿Ļç§į æĥħåĨµä¸ĭ ï¼Į å¦Ĥæŀľ æľºåĻ¨ éĩį åĲ¯ ï¼Į å½ĵåīį command å¯¹åºĶçļĦ èģļ åĲĪ æł¹ ä»İ Event Store æģ ¢å¤į çļĦ èģļ åĲĪ æł¹ æĺ¯ è¢« å½ĵåīį command å¤ĦçĲĨ è¿ĩ åĲİçļĦ ï¼Ľ ĊĠĠĠĠĠĠĠ Ġ// æīĢä»¥ å¦Ĥæŀľ è¯¥ command åĨį æ¬¡ è¢« å¤ĦçĲĨ ï¼Į åı¯èĥ½ å¯¹åºĶçļĦ èģļ åĲĪ æł¹ å°± ä¸įä¼ļ åĨį äº§çĶŁ äºĭä»¶ äºĨ ï¼Ľ ĊĠĠĠĠĠĠĠ Ġ// æīĢä»¥ ï¼Į æĪĳä»¬ è¦ģ èĢĥèĻĳ åĪ° è¿Ļç§į æĥħåĨµ ï¼Į å°Ŀè¯ķ åĨį æ¬¡ åıĳå¸ĥ è¯¥ åĳ½ä»¤ äº§çĶŁ çļĦ äºĭä»¶ åĪ° MQ ï¼Ľ ĊĠĠĠĠĠĠĠ Ġ// åĲ¦åĪĻ ï¼Į å¦Ĥæŀľ æĪĳä»¬ çĽ´æİ¥ å°Ĩ å½ĵåīį command è®¾ç½®ä¸º å®ĮæĪĲ ï¼Į åį³ å¯¹ MQ è¿Ľè¡Į ack æĵįä½ľ ï¼Į éĤ£ è¯¥ command çļĦ äºĭä»¶ å°± æ°¸ è¿ľ ä¸įä¼ļ åĨį åıĳå¸ĥ åĪ° MQ äºĨ ï¼Į è¿Ļæł· å°± æĹłæ³ķ ä¿Ŀè¯ģ CQ RS æķ°æį®çļĦ æľĢç»Ī ä¸Ģèĩ´ æĢ§ äºĨ ãĢĤ ĊĠĠĠĠĠĠĠ Ġif Ġ( dirty Aggregate Root Count Ġ== Ġ 0 Ġ|| Ġchanged Events . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġre publish Command Events ( processing Command , Ġ 0 ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġdirty Aggregate Root !! ĊĠĠĠĠĠĠĠ Ġval Ġevent Stream Ġ= ĠDomain Event Stream ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprocessing Command . message . id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdirty Aggregate Root . unique Id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠtypeName Provider . getType Name ( dirty Aggregate Root . java Class ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠDate (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġchanged Events , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcommand . items ) ĊĠĠĠĠĠĠĠ Ġ// åĨħåŃĺ åħĪ æİ¥åıĹ èģļ åĲĪ æł¹ çļĦ æĽ´æĸ° ï¼Į éľĢè¦ģ æ£ĢæŁ¥ èģļ åĲĪ æł¹ å¼ķçĶ¨ æĺ¯åĲ¦ å·² åıĺåĮĸ ï¼Į å¦Ĥæŀľ å·² åıĺåĮĸ ï¼Į ä¼ļ æĬĽ åĩº å¼Ĥå¸¸ ĊĠĠĠĠĠĠĠ Ġmemory Cache . accept Aggregate Root Changes ( dirty Aggregate Root ) ĊĠĠĠĠĠĠĠ Ġval Ġcommand Result Ġ= Ġprocessing Command . command Execute Context . result ĊĠĠĠĠĠĠĠ Ġprocessing Command . items [ Sys Properties . ITEMS _ COMMAND _ RESULT _ KEY ] Ġ= Ġcommand Result ĊĠĠĠĠĠĠĠ Ġ// æıĲäº¤ äºĭä»¶ æµģ è¿Ľè¡Į åĲİ ç»Ń çļĦ å¤ĦçĲĨ ĊĠĠĠĠĠĠĠ Ġevent Commit ting Service . commit Domain Event Async ( Event Commit ting Context ( event Stream , Ġprocessing Command )) ĊĠĠĠĠĠĠĠ Ġreturn ĠTask . completed Task ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġre publish Command Events ( processing Command : ĠProcessing Command , Ġretry Times : ĠInt ): ĠCompletableFuture < Boolean > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġfuture Ġ= ĠCompletableFuture < Boolean >() ĊĠĠĠĠĠĠĠ Ġval Ġcommand Ġ= Ġprocessing Command . message ĊĠĠĠĠĠĠĠ ĠIO Helper . try Async Action Re cursively (" Process If No Events Of Command ", Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġevent Store . find Async ( command . aggregate Root Id , Ġcommand . id ) ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ Ġresult : ĠDomain Event Stream ? Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( result Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevent Commit ting Service . publish Domain Event Async ( processing Command , Ġresult ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. when Complete Ġ{ Ġ_, Ġ_ Ġ-> Ġfuture . complete ( true ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcomplete Command ( processing Command , ĠCommand Status . Nothing Changed , ĠString :: class . java . name , Ġprocessing Command . command Execute Context . result ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. when Complete Ġ{ Ġ_, Ġ_ Ġ-> Ġfuture . complete ( true ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠString . format ("[ command Id :% s ]", Ġcommand . id ) ĊĠĠĠĠĠĠĠ Ġ}, Ġnull , Ġretry Times , Ġtrue ) ĊĠĠĠĠĠĠĠ Ġreturn Ġfuture ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġhandle Exception Async ( processing Command : ĠProcessing Command , Ġcommand Handler : ĠICommand Handler Proxy , Ġexception : ĠThrowable , ĠerrorMessage : ĠString , Ġretry Times : ĠInt ): ĠCompletableFuture < Boolean > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcommand Ġ= Ġprocessing Command . message ĊĠĠĠĠĠĠĠ Ġval Ġfuture Ġ= ĠCompletableFuture < Boolean >() ĊĠĠĠĠĠĠĠ ĠIO Helper . try Async Action Re cursively (" Find Event By Command Id Async ", Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġevent Store . find Async ( command . aggregate Root Id , Ġcommand . id ) ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ Ġresult : ĠDomain Event Stream ? Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( result Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// è¿ĻéĩĮ ï¼Į æĪĳä»¬ éľĢè¦ģ åĨį éĩįæĸ° åģļ ä¸Ģ éģį åıĳå¸ĥ äºĭä»¶ è¿Ļä¸ª æĵįä½ľ ï¼Ľ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ä¹ĭ æīĢä»¥ è¦ģ è¿Ļæł· åģļ æĺ¯ åĽłä¸º èĻ½çĦ¶ è¯¥ command äº§çĶŁ çļĦ äºĭä»¶ å·²ç»ı æĮģ ä¹ħ åĮĸ æĪĲåĬŁ ï¼Į ä½Ĩ å¹¶ ä¸į è¡¨ç¤º äºĭä»¶ å·²ç»ı åıĳå¸ĥ åĩº åİ» äºĨ ï¼Ľ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// åĽłä¸º æľī åı¯èĥ½ äºĭä»¶ æĮģ ä¹ħ åĮĸ æĪĲåĬŁ äºĨ ï¼Į ä½Ĩ éĤ£ æĹ¶ æŃ£ å¥½ æľºåĻ¨ æĸŃ çĶµ äºĨ ï¼Į åĪĻ åıĳå¸ĥ äºĭä»¶ å°± æ²¡æľī åģļ ï¼Ľ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevent Commit ting Service . publish Domain Event Async ( processing Command , Ġresult ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. when Complete Ġ{ Ġ_, Ġ_ Ġ-> Ġfuture . complete ( true ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// åĪ° è¿ĻéĩĮ ï¼Į è¯´æĺİ å½ĵåīį command æī§è¡Į éģĩ åĪ° å¼Ĥå¸¸ ï¼Į çĦ¶åĲİ å½ĵåīį command ä¹ĭåīį ä¹Ł æ²¡ æī§è¡Į è¿ĩ ï¼Į æĺ¯ ç¬¬ ä¸Ģæ¬¡ è¢« æī§è¡Į ãĢĤ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// éĤ£ å°± åĪ¤æĸŃ å½ĵåīį å¼Ĥå¸¸ æĺ¯åĲ¦ æĺ¯ éľĢè¦ģ è¢« åıĳå¸ĥ åĩº åİ» çļĦ å¼Ĥå¸¸ ï¼Į å¦Ĥæŀľ æĺ¯ ï¼Į åĪĻ åıĳå¸ĥ è¯¥ å¼Ĥå¸¸ ç»Ļ æīĢæľī æ¶Īè´¹ èĢħ ï¼Ľ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// åĲ¦åĪĻ ï¼Į å°± è®°å½ķ éĶĻè¯¯ æĹ¥å¿Ĺ ï¼Į çĦ¶åĲİ è®¤ä¸º è¯¥ command å¤ĦçĲĨ å¤±è´¥ åį³åı¯ ï¼Ľ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġreal Exception Ġ= Ġget Real Exception ( exception ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( real Exception Ġis ĠID omain Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpublish Exception Async ( processing Command , Ġreal Exception Ġas ĠID omain Exception , Ġ 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. when Complete Ġ{ Ġ_, Ġ_ Ġ-> Ġfuture . complete ( true ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcomplete Command ( processing Command , ĠCommand Status . Failed , Ġreal Exception . java Class . name , Ġreal Exception . message ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. when Complete Ġ{ Ġ_, Ġ_ Ġ-> Ġfuture . complete ( true ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠString . format ("[ command :[ id :% s , type :% s ], handler Type :% s , aggregate Root Id :% s ]", Ġcommand . id , Ġcommand . java Class . name , Ġcommand Handler . get Inner Object (). java Class . name , Ġcommand . aggregate Root Id ) ĊĠĠĠĠĠĠĠ Ġ}, Ġnull , Ġretry Times , Ġtrue ) ĊĠĠĠĠĠĠĠ Ġreturn Ġfuture ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġget Real Exception ( exception : ĠThrowable ): ĠThrowable Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( exception Ġis ĠCompletion Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( exception . cause Ġis ĠID omain Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġexception . cause !! ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠArrays . stream ( exception . sup pressed ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġx : ĠThrowable ? Ġ-> Ġx Ġis ĠID omain Exception Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. findFirst () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. orElse ( exception ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġexception ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġpublish Exception Async ( processing Command : ĠProcessing Command , Ġexception : ĠID omain Exception , Ġretry Times : ĠInt ): ĠCompletableFuture < Boolean > Ġ{ ĊĠĠĠĠĠĠĠ Ġexception . merge Items ( processing Command . message . items ) ĊĠĠĠĠĠĠĠ Ġval Ġfuture Ġ= ĠCompletableFuture < Boolean >() ĊĠĠĠĠĠĠĠ ĠIO Helper . try Async Action Re cursively Without Result (" Publish Exception Async ", Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexception Publisher . publish Async ( exception ) ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcomplete Command ( processing Command , ĠCommand Status . Failed , Ġexception . java Class . name , Ġ( exception Ġas ĠException ). message ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. when Complete Ġ{ Ġ_, Ġ_ Ġ-> Ġfuture . complete ( true ) Ġ} ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġserial izable Info : ĠMap < String , ĠAny > Ġ= ĠHashMap () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexception . serialize To ( serial izable Info ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġexception Info Ġ= Ġserial izable Info . entries . join ToString (",") Ġ{ Ġx : ĠMap . Entry < String , ĠAny > Ġ-> ĠString . format ("% s :% s ", Ġx . key , Ġx . value ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠString . format ("[ command Id : Ġ% s , Ġexception Info : Ġ% s ]", Ġprocessing Command . message . id , Ġexception Info ) ĊĠĠĠĠĠĠĠ Ġ}, Ġnull , Ġretry Times , Ġtrue ) ĊĠĠĠĠĠĠĠ Ġreturn Ġfuture ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcommit Changes Async ( processing Command : ĠProcessing Command , Ġsuccess : ĠBoolean , Ġmessage : ĠI Application Message ?, ĠerrorMessage : ĠString ): ĠCompletableFuture < Boolean > Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( success ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( message Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmessage . merge Items ( processing Command . message . items ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġpublish Message Async ( processing Command , Ġmessage , Ġ 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġcomplete Command ( processing Command , ĠCommand Status . Success , Ġ"", Ġ"") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġcomplete Command ( processing Command , ĠCommand Status . Failed , ĠString :: class . java . name , ĠerrorMessage ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġpublish Message Async ( processing Command : ĠProcessing Command , Ġmessage : ĠI Application Message , Ġretry Times : ĠInt ): ĠCompletableFuture < Boolean > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcommand Ġ= Ġprocessing Command . message ĊĠĠĠĠĠĠĠ Ġval Ġfuture Ġ= ĠCompletableFuture < Boolean >() ĊĠĠĠĠĠĠĠ ĠIO Helper . try Async Action Re cursively Without Result (" Publish Application Message Async ", Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġapplication Message Publisher . publish Async ( message ) ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcomplete Command ( processing Command , ĠCommand Status . Success , Ġmessage . java Class . name , Ġserialize Service . serialize ( message )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. when Complete Ġ{ Ġ_, Ġ_ Ġ-> Ġfuture . complete ( true ) Ġ} ĊĠĠĠĠĠĠĠ Ġ}, Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠString . format ("[ application Ġmessage :[ id :% s , type :% s ], command :[ id :% s , type :% s ]] ", Ġmessage . id , Ġmessage . java Class . name , Ġcommand . id , Ġcommand . java Class . name ) ĊĠĠĠĠĠĠĠ Ġ}, Ġnull , Ġretry Times , Ġtrue ) ĊĠĠĠĠĠĠĠ Ġreturn Ġfuture ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġget CommandHandler ( processing Command : ĠProcessing Command , Ġget Handlers Func : ĠFunction < Class <* >, ĠList < MessageHandler Data < IC ommand Handler Proxy >>> ): ĠHandler Find Result Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcommand Ġ= Ġprocessing Command . message ĊĠĠĠĠĠĠĠ Ġval Ġhandler DataList Ġ= Ġget Handlers Func . apply ( command . java Class ) ĊĠĠĠĠĠĠĠ Ġif Ġ( handler DataList . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠHandler Find Result . NotFound ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( handler DataList . size Ġ> Ġ 1 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠHandler Find Result . Too Many Handler Data ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġhandler Data Ġ= Ġhandler DataList . stream (). findFirst (). orElse ( MessageHandler Data ()) ĊĠĠĠĠĠĠĠ Ġif Ġ( handler Data . list Handlers . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠHandler Find Result . NotFound ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( handler Data . list Handlers . size Ġ> Ġ 1 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠHandler Find Result . Too Many Handler ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn ĠHandler Find Result ( Handler Find Status . Found , Ġhandler Data . list Handlers [ 0 ]) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcomplete Command ( processing Command : ĠProcessing Command , Ġcommand Status : ĠCommand Status , Ġresult Type : ĠString , Ġresult : ĠString ?): ĠCompletableFuture < Boolean > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcommand Result Ġ= ĠCommand Result ( command Status , Ġprocessing Command . message . id , Ġprocessing Command . message . aggregate Root Id , Ġresult , Ġresult Type ) ĊĠĠĠĠĠĠĠ Ġreturn Ġprocessing Command . mail Box . complete Message ( processing Command , Ġcommand Result ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġlogger Ġ= ĠLoggerFactory . getLogger ( Default Processing CommandHandler :: class . java ) ĊĠĠĠ Ġ} Ċ }