/* Ċ Ġ* ĠCopyright Ġ 2 0 2 1 ĠThe ĠMatrix . org ĠFoundation ĠC . I . C . Ċ Ġ* Ċ Ġ* ĠLicensed Ġ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 Ġorg . matrix . android . sdk . internal . session . room . send . queue Ċ Ċ import Ġkotlinx . coroutines . Coroutine Scope Ċ import Ġkotlinx . coroutines . Dispatch ers Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . launch Ċ import Ġkotlinx . coroutines . with Context Ċ import Ġorg . matrix . android . sdk . api . auth . data . Session Params Ċ import Ġorg . matrix . android . sdk . api . failure . Failure Ċ import Ġorg . matrix . android . sdk . api . failure . Matrix Error Ċ import Ġorg . matrix . android . sdk . api . failure . get Retry Delay Ċ import Ġorg . matrix . android . sdk . api . session . crypto . Crypto Service Ċ import Ġorg . matrix . android . sdk . api . session . events . model . Event Ċ import Ġorg . matrix . android . sdk . api . util . Cancel able Ċ import Ġorg . matrix . android . sdk . internal . session . Session Scope Ċ import Ġorg . matrix . android . sdk . internal . task . Coroutine Se qu encer Ċ import Ġorg . matrix . android . sdk . internal . task . Semaphore Coroutine Se qu encer Ċ import Ġorg . matrix . android . sdk . internal . task . Task Executor Ċ import Ġorg . matrix . android . sdk . internal . util . to Cancel able Ċ import Ġtim ber . log . Tim ber Ċ import Ġjava . io . IOException Ċ import Ġjava . util . concurrent . Concurrent HashMap Ċ import Ġjava . util . concurrent . atomic . Atomic Boolean Ċ import Ġjavax . inject . Inject Ċ import Ġkotlin . coroutines . c ancellation . Cancellation Exception Ċ Ċ private Ġconst Ġval ĠRET RY _ WAIT _ TIME _ MS Ġ= Ġ 1 0 _ 0 0 0 L Ċ private Ġconst Ġval ĠMAX _ RETRY _ COUNT Ġ= Ġ 3 Ċ Ċ /** Ċ Ġ* ĠThis Ġclass Ġis Ġresponsible Ġfor Ġsending Ġevents Ġin Ġorder Ġin Ġeach Ġroom . ĠIt Ġuses Ġthe ĠQue ued Task . queue Identifier Ġto Ġexecute Ġtasks Ġse quent ially . Ċ Ġ* ĠEach Ġsend Ġis Ġretried Ġ 3 Ġtimes , Ġif Ġthere Ġis Ġno Ġnetwork Ġ( e . g Ġif Ġcannot Ġping Ġhome Ġserver ) Ġit Ġwill Ġwait Ġand Ċ Ġ* Ġperiodically Ġtest Ġreach ability Ġbefore Ġresume Ġ( does Ġnot Ġcount Ġas Ġa Ġretry ) Ċ Ġ* Ċ Ġ* ĠIf Ġthe Ġapp Ġis Ġkilled Ġbefore Ġall Ġevent Ġwere Ġsent , Ġon Ġnext Ġwake up Ġthe Ġscheduled Ġevents Ġwill Ġbe Ġre Ġposted Ċ Ġ* Ċ Ġ*/ Ċ @ Session Scope Ċ internal Ġclass ĠEvent Sender Processor Coroutine Ġ@ Inject Ġconstructor ( ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġcrypto Service : ĠCrypto Service , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġsession Params : ĠSession Params , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġqueued Task Factory : ĠQue ued Task Factory , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġtask Executor : ĠTask Executor , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġme ment o : ĠQueue Mem ento Ċ ) Ġ: ĠEvent Sender Processor Ġ{ ĊĊĠĠĠ Ġprivate Ġval ĠwaitFor Network Se qu encer Ġ= ĠSem aphore Coroutine Se qu encer () ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġsequ enc ers Ġuse ĠQue ued Task . queue Identifier Ġas Ġkey ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġval Ġsequ enc ers Ġ= ĠConcurrent HashMap < String , ĠCor outine Se qu encer >() ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġcancel able Bag Ġuse ĠQue ued Task . task Identifier Ġas Ġkey ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġval Ġcancel able Bag Ġ= ĠConcurrent HashMap < String , ĠCancel able >() ĊĊĠĠĠ Ġoverride Ġfun Ġon Session Started () Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠWe Ġshould Ġcheck Ġfor Ġsending Ġevents Ġnot Ġhandled Ġbecause Ġapp Ġwas Ġkilled ĊĠĠĠĠĠĠĠ Ġ// ĠBut Ġwe Ġshould Ġbe Ġcareful Ġof Ġonly Ġtook Ġthose Ġthat Ġwas Ġsubmitted Ġto Ġus , Ġbecause Ġif Ġit 's ĊĠĠĠĠĠĠĠ Ġ// Ġfor Ġexample Ġit 's Ġa Ġmedia Ġevent Ġit Ġis Ġhandled Ġby Ġsome Ġworker Ġand Ġhe Ġwill Ġhandle Ġit ĊĠĠĠĠĠĠĠ Ġ// ĠThis Ġis Ġa Ġbit Ġfrag ile Ġ: / ĊĠĠĠĠĠĠĠ Ġ// Ġalso Ġsome Ġevents Ġcannot Ġbe Ġretried Ġmanually Ġby Ġusers , Ġe . g Ġre actions ĊĠĠĠĠĠĠĠ Ġ// Ġthey Ġwere Ġpreviously Ġrelying Ġon Ġworkers Ġto Ġdo Ġthe Ġwork Ġ: / Ġand Ġwas Ġexpected Ġto Ġalways Ġfinally Ġsucceed ĊĠĠĠĠĠĠĠ Ġ// ĠAlso Ġsome Ġe ch os Ġare Ġnot Ġto Ġbe Ġres ent Ġlike Ġred action Ġe ch os Ġ( fake Ġevent Ġcreated Ġfor Ġaggregation ) ĊĠĠĠĠĠĠĠ Ġtask Executor . executor Scope . launch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠTim ber . d (" ## ĠSend Ġrel a unched Ġpending Ġevents Ġon Ġrestart ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġme ment o . restore Tasks ( this @ Event Sender Processor Coroutine ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( failure : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTim ber . e ( failure , Ġ" Fail Ġrest oring Ġsend Ġtasks ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġpost Event ( event : ĠEvent ): ĠCancel able Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġpost Event ( event , Ġevent . roomId ?. let Ġ{ Ġcrypto Service . is Room Encrypted ( it ) Ġ} Ġ?: Ġfalse ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġpost Event ( event : ĠEvent , Ġencrypt : ĠBoolean ): ĠCancel able Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtask Ġ= Ġqueued Task Factory . create Send Task ( event , Ġencrypt ) ĊĠĠĠĠĠĠĠ Ġreturn Ġpost Task ( task ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġpost Red action ( red action Local Echo : ĠEvent , Ġreason : ĠString ?): ĠCancel able Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġpost Red action ( red action Local Echo . eventId !! , Ġred action Local Echo . red acts !! , Ġred action Local Echo . roomId !! , Ġreason ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġpost Red action ( red action Local Echo Id : ĠString , Ġevent To Red act Id : ĠString , Ġroom Id : ĠString , Ġreason : ĠString ?): ĠCancel able Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtask Ġ= Ġqueued Task Factory . create Red act Task ( red action Local Echo Id , Ġevent To Red act Id , Ġroom Id , Ġreason ) ĊĠĠĠĠĠĠĠ Ġreturn Ġpost Task ( task ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġpost Task ( task : ĠQue ued Task ): ĠCancel able Ġ{ ĊĠĠĠĠĠĠĠ Ġmark As Managed ( task ) ĊĠĠĠĠĠĠĠ Ġval Ġsequ encer Ġ= Ġsequ enc ers . get Or Put ( task . queue Identifier ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠSem aphore Coroutine Se qu encer () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ ĠTim ber . v (" ## Ġpost Ġ$ task ") ĊĠĠĠĠĠĠĠ Ġreturn Ġtask Executor . executor Scope ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. launch With ( sequ encer ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġexecute Task ( task ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}. to Cancel able () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. also Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcancel able Bag [ task . task Identifier ] Ġ= Ġit ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġcancel ( eventId : ĠString , Ġroom Id : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠ Ġ// Ġevent Id Ġis Ġmost Ġlikely Ġthe Ġtask Identifier ĊĠĠĠĠĠĠĠ Ġcancel able Bag [ eventId ]?. cancel () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun ĠCor outine Scope . launch With ( sequ encer : ĠCor outine Se qu encer , Ġblock : Ġsuspend ĠCor outine Scope . () Ġ-> ĠUnit ) Ġ= Ġlaunch Ġ{ ĊĠĠĠĠĠĠĠ Ġsequ encer . post Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġblock () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġexecute Task ( task : ĠQue ued Task ) Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( task . is Cancelled ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTim ber . v (" ## Ġ$ task Ġhas Ġbeen Ġcancelled , Ġtry Ġnext Ġtask ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtask . waitFor Network () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtask . execute () ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( exception : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġexception Ġis ĠIOException Ġ|| Ġexception Ġis ĠFailure . Network Connection ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcan Reach Server . set ( false ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtask . mark As Failed Or Retry ( exception , Ġ 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ( exception Ġis ĠFailure . ServerError Ġ&& Ġexception . error . code Ġ== ĠMatrix Error . M _ LIMIT _ EX CE EDED ) Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtask . mark As Failed Or Retry ( exception , Ġexception . get Retry Delay ( 3 _ 0 0 0 )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġexception Ġis ĠC ancellation Exception ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTim ber . v (" ## Ġ$ task Ġhas Ġbeen Ġcancelled , Ġtry Ġnext Ġtask ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTim ber . v (" ## Ġun - retry able Ġerror Ġfor Ġ$ task , Ġtry Ġnext Ġtask ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthis Ġtask Ġis Ġin Ġerror , Ġcheck Ġnext Ġone ? ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtask . on Task Failed () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġmark As Finished ( task ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun ĠQue ued Task . mark As Failed Or Retry ( failure : ĠThrowable , Ġretry Delay : ĠLong ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( retry Count . increment AndGet () Ġ>= ĠMAX _ RETRY _ COUNT ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Task Failed () ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠTim ber . v (" ## Ġretry able Ġerror Ġfor Ġ$ this Ġreason : Ġ${ failure . localized Message }") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWait Ġif Ġnecessary ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( retry Delay ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠAnd Ġthen Ġretry ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexecute Task ( this ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġmark As Managed ( task : ĠQue ued Task ) Ġ{ ĊĠĠĠĠĠĠĠ Ġme ment o . track ( task ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġmark As Finished ( task : ĠQue ued Task ) Ġ{ ĊĠĠĠĠĠĠĠ Ġcancel able Bag . remove ( task . task Identifier ) ĊĠĠĠĠĠĠĠ Ġme ment o . un Track ( task ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġval Ġcan Reach Server Ġ= ĠAtomic Boolean ( true ) ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun ĠQue ued Task . waitFor Network () Ġ= ĠwaitFor Network Se qu encer . post Ġ{ ĊĠĠĠĠĠĠĠ Ġwhile Ġ(! can Reach Server . get ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠTim ber . v (" ## Ġ$ this Ġcannot Ġreach Ġserver Ġwait Ġts :${ System . currentTimeMillis () }") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( RETRY _ WAIT _ TIME _ MS ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwith Context ( Dispatch ers . IO ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġhost Available Ġ= ĠHome Server Availability Checker ( session Params ). check () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcan Reach Server . set ( host Available ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ