package Ġcom . lab ij ie . infra . mq ts Ċ Ċ import Ġcom . lab ij ie . infra . I Id Generator Ċ import Ġcom . lab ij ie . infra . impl . Debug Id Generator Ċ import Ġio . netty . util . Has hed Wheel Timer Ċ import Ġcom . lab ij ie . infra . mq ts . impl . Default Transaction Holder Ċ import Ġcom . lab ij ie . infra . mq ts . impl . Jack son Data Serializer Ċ import Ġcom . lab ij ie . infra . mq ts . impl . Memory Transaction Queue Ċ import Ġcom . lab ij ie . infra . mq ts . impl . Simple Instance Factory Ċ import Ġcom . lab ij ie . infra . mq ts . ab stractions .* Ċ import Ġcom . lab ij ie . infra . mq ts . ack . Ack Request Ċ import Ġcom . lab ij ie . infra . mq ts . ack . Ack RequestContext Ċ import Ġcom . lab ij ie . infra . mq ts . ack . Callback Key Ċ import Ġcom . lab ij ie . infra . mq ts . configuration . MQ Transaction Config Ċ import Ġcom . lab ij ie . infra . mq ts . context . Transaction Context Ċ import Ġcom . lab ij ie . infra . utils . throw If Ne cessary Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġjava . lang . reflect . Invocation TargetException Ċ import Ġjava . net . URI Ċ import Ġjava . util .* Ċ import Ġjava . util . concurrent .* Ċ import Ġjava . util . concurrent . atomic . Atomic Long Ċ import Ġkotlin . collections . HashSet Ċ import Ġkotlin . reflect . K Class Ċ import Ġkotlin . reflect . full . is Subclass Of ĊĊ Ċ class ĠMQ Transaction Manager Ġconstructor ( ĊĠĠĠĠĠĠĠ Ġval Ġide mpot ence : ĠII de mpot ence , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġrepository : ĠI Transaction Repository , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġdata Serializer : ĠI Transaction Data Serializer Ġ= ĠJack son Data Serializer (), ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġid Generator : ĠI Id Generator Ġ= ĠDebug Id Generator (), ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġconfig : ĠMQ Transaction Config Ġ= ĠMQ Transaction Config (), ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġinstance Factory : ĠI Instance Factory Ġ= ĠSimple Instance Factory (), ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġtransaction Holder : ĠI Transaction Holder Ġ= ĠDefault Transaction Holder (), ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġm q : ĠI Transaction Queue Ġ= ĠMemory Transaction Queue ( transaction Holder ), ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġhost : ĠString Ġ= Ġ"") Ġ{ ĊĊĠĠĠ Ġval Ġtransaction Listeners Ġ= ĠTransaction Listeners () ĊĊĠĠĠ Ġprivate Ġval Ġsource Sync : ĠAny Ġ= ĠAny () ĊĠĠĠ Ġprivate Ġval Ġparticipant Sync : ĠAny Ġ= ĠAny () ĊĠĠĠ Ġprivate Ġval Ġretry Timer : ĠHas hed Wheel Timer ĊĠĠĠ Ġprivate Ġval Ġtransaction Attributes Ġ= ĠHashMap < String , ĠTransaction Source Attribute >() ĊĊĠĠĠ Ġprivate Ġval Ġsource Attributes Ġ= ĠHashMap < Transaction Identifier , ĠMutable Map < String , ĠTransaction Source Attribute >>() ĊĠĠĠ Ġprivate Ġval Ġparticipant Attributes Ġ= ĠHashMap < Transaction Identifier , ĠMutable Map < String , ĠTransaction Participant Attribute >>() ĊĠĠĠ Ġprivate Ġval Ġcallback Set Ġ= ĠHashMap < Callback Key , ĠHashSet < IT rans action Ack Callback >>() ĊĊĠĠĠ Ġprivate Ġval Ġretry Tasks : ĠConcurrent HashMap < String , ĠConcurrent HashMap < Long , ĠTransaction Pro ducing Context >> Ġ= ĠConcurrent HashMap () ĊĠĠĠ Ġprivate Ġval Ġretry Thread Count Ġ= ĠAtomic Long ( 0 ) ĊĊĊĠĠĠ Ġval Ġtransaction Sources : ĠCollection < Transaction Source Attribute > ĊĠĠĠĠĠĠĠ Ġget () Ġ= Ġthis . transaction Attributes . values ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġs Ġ= ĠSystem . get Security Manager () ĊĠĠĠĠĠĠĠ Ġval Ġgroup Ġ= Ġif Ġ( s Ġ!= Ġnull ) Ġs . thread Group Ġelse ĠThread . currentThread (). thread Group ĊĊĠĠĠĠĠĠĠ Ġthis . retry Timer Ġ= ĠHas hed Wheel Timer ({ Ġr Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ ĠThread ( group , Ġr , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" mq ts - retry -${ retry Thread Count . increment AndGet () }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( this . is Daemon ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . is Daemon Ġ= Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( this . priority Ġ!= ĠThread . NORM _ PRIORITY ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . priority Ġ= ĠThread . NORM _ PRIORITY ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}, Ġ 1 , ĠTimeUnit . SECONDS ) ĊĊĠĠĠĠĠĠĠ ĠRuntime . get Runtime (). add Shutdown Hook ( Thread Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . retry Timer . stop () ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġ// val ĠP ARENT _ TRANSACTION _ STATE Ġ= Ġ"__ parent " ĊĠĠĠĠĠĠĠ Ġconst Ġval ĠP AY LOAD _ STATE _ PREFIX Ġ= Ġ"__ pl :" ĊĠĠĠĠĠĠĠ Ġval ĠDEFAULT _ RETRY _ T IC KS Ġ= Ġint ArrayOf ( 5 , Ġ 1 0 , Ġ 1 0 , Ġ 3 0 , Ġ 6 0 ) ĊĊĠĠĠĠĠĠĠ Ġval Ġlogger Ġ= ĠLoggerFactory . getLogger ( MQ Transaction Manager :: class . java ) ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġ@ Throws ( MQ Transaction Exception :: class ) ĊĠĠĠ Ġfun Ġfind Participant ( identifier : ĠTransaction Identifier ): ĠMap < String , ĠTransaction Participant Attribute > Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ(! this . participant Attributes . containsKey ( identifier )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġmap Of () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġthis . participant Attributes [ identifier ] !!. toMap () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Throws ( MQ Transaction Exception :: class ) ĊĠĠĠ Ġfun Ġfind Source ( identifier : ĠTransaction Identifier ): ĠMap < String , ĠTransaction Source Attribute > Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ(! this . source Attributes . containsKey ( identifier )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġmutable Map Of () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġthis . source Attributes [ identifier ] !!. toMap () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġregister Listeners ( var arg Ġlistener : ĠI Transaction Listener ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( listener . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . transaction Listeners . addAll ( listener ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġstop All Retry () Ġ{ ĊĠĠĠĠĠĠĠ Ġthis . retry Timer . stop () ĊĠĠĠĠĠĠĠ Ġthis . retry Tasks . values . forEach Ġ{ Ġtasks Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtasks . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . value . cancel () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġthis . retry Tasks . clear () ĊĠĠĠĠĠĠĠ Ġlogger . info (" MQ Ġtransaction Ġmanager Ġwas Ġshutdown .") ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġ@ Throws ( MQ Transaction Exception :: class ) ĊĠĠĠ Ġprivate Ġfun Ġget Transaction Attribute ( type : ĠString , Ġthrow If NotFound : ĠBoolean Ġ= Ġtrue ): ĠTransaction Source Attribute ? Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( source Attributes . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( throw If NotFound ) Ġthrow ĠMQ Transaction Exception (" cant Ġfind Ġany Ġtransaction Ġattributes ") Ġelse Ġlogger . warn (" cant Ġfind Ġtransaction Ġtype Ġ'$ type ' Ġin Ġcode , Ġbut Ġit Ġwas Ġin Ġtransaction Ġstorage ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġattribute Ġ= Ġthis . transaction Attributes [ type ] ĊĠĠĠĠĠĠĠ Ġif Ġ( attribute Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( throw If NotFound ) Ġthrow ĠMQ Transaction Exception (" cant Ġfind Ġtransaction Ġtype Ġ'$ type '") Ġelse Ġlogger . warn (" cant Ġfind Ġtransaction Ġtype Ġ'$ type ' Ġin Ġcode , Ġbut Ġit Ġwas Ġin Ġtransaction Ġstorage ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġattribute ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġ åĲ¯çĶ¨ äºĭåĬ¡ æºĲ ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġenable Source ( var arg Ġsources : ĠTransaction Source Attribute ) Ġ{ ĊĠĠĠĠĠĠĠ Ġsynchronized ( this . source Sync ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( sources . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . retry Timer . start () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtype Names Ġ= ĠHashSet < String >() ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsources . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġidentifier Ġ= ĠTransaction Identifier ( it . declare Class , Ġit . method ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmap Ġ= Ġthis . source Attributes . get Or Put ( identifier ) Ġ{ Ġmutable Map Of < String , ĠTransaction Source Attribute >() Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmap . put If Absent ( it . annotation . type , Ġit ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( it . annotation . ack Callback Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcallback Ġ= Ġit . annotation . ack Callback !! ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! callback . is Subclass Of ( IT rans action Ack Callback :: class )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠMQ Transaction Exception (" The Ġack Callback Ġproperty Ġof ĠMQ Transaction Source Ġannotation Ġmust Ġbe Ġa ĠI Transaction Ack Callback Ġtype ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ç±»åŀĭ åĴĮ åĽŀè°ĥ ç±»åŀĭ åİ» éĩį ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( it . annotation . ack Callback Ġ!= Ġnull Ġ&& Ġtype Names . add ("${ it . annotation . type }_${ callback . simple Name } ")) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcallback Instance Ġ= Ġthis . instance Factory . create Instance ( it . annotation . ack Callback !! ) Ġas ĠI Transaction Ack Callback ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġset Ġ= Ġcallback Set . get Or Put ( Callback Key ( it . annotation . queue , Ġit . annotation . type )) Ġ{ ĠHashSet () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset . add ( callback Instance ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( this . transaction Attributes . containsKey ( it . annotation . type )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġexisted Ġ= Ġthis . transaction Attributes [ it . annotation . type ] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( exist ed Ġ!= Ġnull Ġ&& Ġexisted . config Equals ( it )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" Transaction Ġwith Ġtype Ġ'${ it . annotation . type }' Ġused Ġdifferent Ġconfiguration ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . transaction Attributes [ it . annotation . type ] Ġ= Ġit ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġfun Ġrecovery Transactions ( ack Server : ĠI Ack Server ): ĠRecovery Result Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġre covered Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġexpired Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtran Ġ= Ġrepository . get Available () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtran . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// å¦Ĥæŀľ æ²¡æľī è¿ĩ æľŁ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsource Attribute Ġ= Ġget Transaction Attribute ( it . transaction Type , Ġthrow If NotFound Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( source Attribute Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġscope Context Ġ= ĠTransaction Context ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! it . is Expired ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . beginTransaction ( ack Server , Ġscope Context , Ġis Reload Ġ= Ġtrue ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġre covered Ġ++ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . expire Transaction ( scope Context ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" Transaction Ġ(' ${ it . transaction Id }') Ġhas Ġexpired ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġexpired ++ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠRecovery Result ( re covered , Ġexpired ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" An Ġerror Ġoccurred Ġwhile Ġinitializing Ġm q Ġtransaction Ġsource ", Ġex ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġex ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġ åĲ¯çĶ¨ äºĭåĬ¡ åıĤ ä¸İ èĢħ ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġenable Participant ( var arg Ġparticipants : ĠTransaction Participant Attribute ) Ġ{ ĊĠĠĠĠĠĠĠ Ġsynchronized ( this . participant Sync ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġparticipants . forEach Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġidentifier Ġ= ĠTransaction Identifier ( it . declare Class , Ġit . method ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmap Ġ= Ġthis . participant Attributes . get Or Put ( identifier ) Ġ{ Ġmutable Map Of () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmap . put If Absent ( it . info . type , Ġit ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( it . method . parameters . size Ġ! in ĠlistOf ( 1 , Ġ 2 )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠMQ Transaction Exception (" transaction Ġparticipant Ġmethod Ġcan Ġcontain Ġonly Ġone Ġparameter Ġor Ġzero Ġparameter .") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġinstance Ġ= Ġthis . instance Factory . create Instance ( it . declare Class ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġm q . register Handler ( it . info . queue , Ġit . info . type ) Ġ{ Ġm qt Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( mq t . transaction Type . compareTo ( it . info . type , Ġtrue ) Ġ== Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġscope Context Ġ= ĠTransaction Context ( mq t ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . transaction Listeners . on Transaction Executing ( scope Context ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( mq t . data Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( it . method . parameters . size Ġ== Ġ 2 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠMQ Transaction Exception (" Transaction Ġparticipant Ġmethod Ġwithout Ġparameter , Ġbut Ġgot Ġone .") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . method . call ( instance ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( it . method . parameters . size Ġ== Ġ 1 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠMQ Transaction Exception (" Transaction Ġparticipant Ġmethod Ġneed Ġat Ġleast Ġone Ġparameter .") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval ĠdataType Ġ= Ġ( it . method . parameters [ 1 ]. type . classifier Ġas ĠK Class < out ĠAny >) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdata : ĠAny Ġ= Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . data Serializer . deserialize ( mq t . data !! , ĠdataType ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġscope Context . exception Ġ= Ġex ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Input Ġdata Ġof Ġparticipant Ġcannot Ġbe Ġdeserialized , Ġconsidering Ġthat Ġthe Ġinput Ġand Ġoutput Ġdata Ġtypes Ġof Ġtransaction Ġsource Ġand Ġparticipant Ġdo Ġnot Ġmatch Ġ( Ġtransaction - type :${ it . info . type } Ġ) Ġ. ", Ġex ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġex . throw If Ne cessary () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ run ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . method . call ( instance , Ġdata ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠInvocation TargetException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( ex . cause Ġ! is ĠIde mpot ence Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġscope Context . exception Ġ= Ġif Ġ( ex . cause Ġ!= Ġnull ) Ġex . cause Ġelse Ġex ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġex ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġscope Context . exception Ġ= Ġex ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġex ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . transaction Listeners . on Transaction Executed ( scope Context ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġfun ĠgetCurrent Transaction (): ĠMQ Transaction ? Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġthis . transaction Holder . current Transaction ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġfind Transaction ( transaction Id : ĠLong ): ĠMQ Transaction ? Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġthis . repository . getById ( transaction Id ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġrecover Expired Transaction ( transaction Id : ĠLong , Ġtransaction Type : ĠString ): ĠMQ Transaction ? Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġsource Ġ= Ġget Transaction Attribute ( transaction Type , Ġtrue ) !! ĊĠĠĠĠĠĠĠ Ġreturn Ġthis . repository . recover Expired Transaction ( transaction Id , Ġsource . annotation . timeout Seconds ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġprepare New Transaction ( source : ĠI Source Info , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdata : ĠByteArray ? Ġ= Ġnull , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġparent Transaction : ĠMQ Transaction ? Ġ= Ġnull , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠreturnValue : ĠAny ?): ĠTransaction Context Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġid Ġ= Ġthis . id Generator . new Id () ĊĠĠĠĠĠĠĠ Ġval Ġnow Ġ= ĠSystem . currentTimeMillis () ĊĊĠĠĠĠĠĠĠ Ġval Ġm q Transaction Ġ= ĠMQ Transaction ( id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsource . type , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnow , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnow Ġ+ Ġsource . timeout Seconds Ġ* Ġ 1 0 0 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdata , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġhost , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġparent Transaction ?. transaction Id ) ĊĊĊĠĠĠĠĠĠĠ Ġval Ġpayload Ġ= ĠreturnValue Ġas ? ĠI MQ Transaction Payload ĊĊĠĠĠĠĠĠĠ Ġpayload ?. extract ()?. forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġm q Transaction . states [" $ PAYLOAD _ STATE _ PREFIX ${ it . key } "] Ġ= Ġit . value ĊĠĠĠĠĠĠĠ Ġ} Ċ Ċ // ĠĠĠĠĠĠĠ Ġif Ġ( parent Transaction Ġ!= Ġnull ) Ġ{ Ċ // ĠĠĠĠĠĠĠĠĠĠĠ Ġm q Transaction . states [ PARENT _ TRANSACTION _ STATE ] Ġ= ĠObjectMapper (). writeValue AsString ( parent Transaction ) Ċ // ĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġscope Context Ġ= ĠTransaction Context ( mq Transaction , Ġparent Transaction , ĠreturnValue ) ĊĠĠĠĠĠĠĠ Ġthis . transaction Listeners . on Transaction Prepared ( scope Context ) ĊĠĠĠĠĠĠĠ Ġthis . repository . save ( mq Transaction , Ġsource ) ĊĠĠĠĠĠĠĠ Ġreturn Ġscope Context ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġ å¼Ģå§ĭ ä¸Ģä¸ª äºĭåĬ¡ ĊĠĠĠĠ Ġ* Ġ@ param Ġack Server Ġ è´Ł è´£ äºĭåĬ¡ Ġack Ġ åºĶ çŃĶ çļĦ æľįåĬ¡åĻ¨ ĊĠĠĠĠ Ġ* Ġ@ param Ġtransaction Scope Context Ġ è¦ģ å¼Ģå§ĭ çļĦ äºĭåĬ¡ å¯¹è±¡ ĊĠĠĠĠ Ġ* Ġ@ param Ġsource Attribute Ġ äºĭåĬ¡ çĽ¸åħ³ çļĦ å±ŀæĢ§ ï¼Ī åħĥ æķ°æį® ï¼ī ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġbegin Transaction ( ack Server : ĠI Ack Server , Ġtransaction Scope Context : ĠTransaction Context , Ġsource Attribute : ĠTransaction Source Attribute ? Ġ= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠ Ġthis . beginTransaction ( ack Server , Ġtransaction Scope Context , Ġsource Attribute , Ġfalse ) ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġprivate Ġfun Ġthrow If OutOf Size ( transaction : ĠMQ Transaction ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( transaction . data Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( transaction . data !!. size Ġ> Ġthis . config . max DataSize Bytes ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠMQ Transaction Exception (" Mq ts Ġdata Ġexceeded Ġlimit Ġ( Ġmax Ġsize : Ġ${ config . max DataSize Bytes } Ġbytes )") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġbegin Transaction ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġack Server : ĠI Ack Server , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtransaction Scope Context : ĠTransaction Context , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsource Attribute : ĠTransaction Source Attribute ? Ġ= Ġnull , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis Reload : ĠBoolean Ġ= Ġfalse ) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġval Ġtransaction Ġ= Ġtransaction Scope Context . transaction ĊĠĠĠĠĠĠĠ Ġthis . throw If OutOf Size ( transaction ) ĊĊĠĠĠĠĠĠĠ Ġ// æģ ¢å¤į äºĭåĬ¡ æĹ¶åĢĻ åı¯èĥ½ åıĳçĶŁ æºĲ åľ°åĿĢ åıĺåĮĸ ĊĠĠĠĠĠĠĠ Ġtransaction . ack Host And Port Ġ= Ġack Server . get Ack Address (). toString () ĊĊĠĠĠĠĠĠĠ Ġvar Ġsource Ġ= Ġsource Attribute ĊĠĠĠĠĠĠĠ Ġif Ġ( source Attribute Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsource Ġ= Ġget Transaction Attribute ( transaction . transaction Type ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ// retry Timer . new Timeout () ĊĠĠĠĠĠĠĠ Ġval Ġcontext Ġ= ĠTransaction Pro ducing Context ( this . mq , Ġtransaction Scope Context . transaction , Ġsource !! , Ġ! is Reload , Ġtransaction Scope Context . states ) ĊĊĠĠĠĠĠĠĠ Ġval Ġtask Map Ġ= Ġthis . retry Tasks . get Or Put ( context . transaction . transaction Type ) Ġ{ ĠConcurrent HashMap () Ġ} ĊĠĠĠĠĠĠĠ Ġif Ġ( task Map . put If Absent ( context . transaction . transaction Id , Ġcontext ) Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! this . is Transaction Exist ed ( context . transaction . transaction Id )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . remove Transaction Retry Task ( transaction Scope Context . transaction ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Transaction Ġwith Ġid Ġ'${ context . transaction . transaction Id }' Ġwas Ġready Ġto Ġbegin , Ġbut Ġit Ġwas Ġnot Ġexisted .") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġexecution Ġ= ĠTransaction Producer ( this , Ġthis . retry Timer , Ġcontext , Ġthis . transaction Listeners ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexecution . execute () ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" C ant Ġto Ġbegin Ġa Ġtransaction Ġ'${ context . transaction . transaction Id }' Ġthat Ġexisted Ġin Ġretry Ġtask .") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġ å®ĮæĪĲ ä¸Ģä¸ª äºĭåĬ¡ ï¼Ī æ¸ħ çĲĨ éĩį è¯ķ ä»»åĬ¡ ï¼Į å¹¶ æıĲäº¤ æķ°æį® ï¼ī ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġfun Ġcommit Transaction ( context : ĠTransaction Context ) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! this . repository . delete By Transaction Id ( context . transaction . transaction Id )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcontext . ide mpot ent Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug (" Redo Ġcomplete Ġtransaction .") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ide : ĠIde mpot ence Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcontext . ide mpot ent Ġ= Ġtrue ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcontext . exception Ġ= Ġex ĊĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! remove Transaction Retry Task ( context . transaction )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug (" MQ TS Ġretry Ġtask Ġcancel Ġfailed , Ġpossibly Ġbecause Ġthe Ġtask Ġhas Ġstarted Ġ( Ġtran - id :${ context . transaction . transaction Id }, Ġtype :${ context . transaction . transaction Type } Ġ) ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug (" MQ TS Ġretry Ġtask Ġwas Ġcanceled .") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . transaction Listeners . on Transaction Completed ( context ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġinternal Ġfun Ġremove Transaction Retry Task ( transaction : ĠMQ Transaction ): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġset Ġ= Ġthis . retry Tasks . get Or Put ( transaction . transaction Type ) Ġ{ ĠConcurrent HashMap () Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġp ctx Ġ= Ġset . remove ( transaction . transaction Id ) ĊĠĠĠĠĠĠĠ Ġp ctx ?. cancel () ĊĠĠĠĠĠĠĠ Ġreturn Ġp ctx Ġ!= Ġnull ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġexpire Transaction ( transaction Scope Context : ĠTransaction Context ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( this . repository . expire By Transaction Id ( transaction Scope Context . transaction . transaction Id )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . transaction Listeners . on Transaction Expired ( transaction Scope Context ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġremove Transaction Retry Task ( transaction Scope Context . transaction ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġis Transaction Exist ed ( transaction Id : ĠLong ): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . repository . is Transaction Exist ed ( transaction Id ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// åĩº éĶĻ æĹ¶åĢĻ å®ģ æĦ ¿ çĽ¸ ä¿¡ äºĭåĬ¡ åŃĺåľ¨ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" Execute Ġis Transaction Exist ed Ġfault .", Ġex ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġtransaction Scope Context From Ack Request ( transaction : ĠMQ Transaction , Ġrequest : ĠAck Request ): ĠTransaction Context Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtran Ctx Ġ= ĠTransaction Context ( transaction ) ĊĠĠĠĠĠĠĠ Ġrequest . transaction States . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtran Ctx . transaction . states [ it . key ] Ġ= Ġit . value ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġtran Ctx ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġcomplete Transaction ( queue : ĠString , Ġack Request : ĠAck Request , Ġresult : ĠTransaction Result ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtransaction Ġ= Ġthis . find Transaction ( ack Request . transaction Id ) ĊĠĠĠĠĠĠĠ Ġif Ġ( transaction Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtran Ctx Ġ= Ġtransaction Scope Context From Ack Request ( transaction , Ġack Request ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . transaction Listeners . on Transaction Ack Received ( tran Ctx ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcontext Ġ= ĠAck Context ( transaction , Ġthis , Ġresult ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthis . commit Transaction ( tran Ctx ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcallbacks Ġ= Ġthis . callback Set . get OrDefault ( Callback Key ( queue , Ġtransaction . transaction Type ), Ġnull ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( callbacks Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcallbacks . forEach Ġ{ Ġc Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// æīĭåĬ¨ æıĲäº¤ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġc . on Ack ( context ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠIde mpot ence Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" Redo Ġcomplete Ġtransaction Ġ( Ġtransaction Ġid : Ġ${ transaction . transaction Id } Ġ) ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġex ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ Ġ// æ²¡æľī éħįç½® åĽŀè°ĥ çļĦ æĥħåĨµä¸ĭ èĩªåĬ¨ æıĲäº¤ äºĭåĬ¡ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcontext . commit MQ Transaction () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠIde mpot ence Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" Redo Ġcomplete Ġtransaction Ġ( Ġtransaction Ġid : Ġ${ transaction . transaction Id } Ġ) ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġex ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! context . is Commit ted ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" Transaction Ġwill Ġnot Ġbe Ġcompleted Ġunless ĠAck Context . commit Transaction Ġmethod Ġinvoked .") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġ åºĶ çŃĶ ä¸Ģä¸ª äºĭåĬ¡ çļĦ æī§è¡Į ç»ĵæŀľ ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġ@ Throws ( MQ Transaction Exception :: class ) ĊĠĠĠ Ġfun Ġack Transaction ( transaction : ĠMQ Transaction , Ġack Client : ĠI Ack Client , Ġparticipant : ĠI Participant Info , Ġresult : ĠTransaction Result ) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġval Ġcontext Ġ= ĠTransaction Context ( transaction ) ĊĊĠĠĠĠĠĠĠ Ġtransaction Listeners . on Transaction Ack Request ing ( context ) ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġack Context Ġ= ĠAck RequestContext ( URI ( transaction . ack Host And Port ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtransaction . transaction Id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtransaction . transaction Type , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġparticipant . queue , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtransaction . extract States Without Payload ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġack Client . write ( ack Context , Ġresult ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" MQ Ġtransaction Ġinvoking Ġack Ġserver Ġerror .", Ġex ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcontext . exception Ġ= Ġex ĊĠĠĠĠĠĠĠĠĠĠĠ Ġex . throw If Ne cessary () ĊĊĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtransaction Listeners . on Transaction Ack Requested ( context ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ }