/* Ċ Ġ* ĠCopyright ĠÂ© Ġ 2 0 2 1 ĠPaul ĠAmb rose Ġ( p amb rose @ mac . com ) Ċ Ġ* Ċ Ġ* Ġ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 . Ċ Ġ*/ Ċ Ċ @ file : Suppress (" Un document ed Public Class ", Ġ" Un document ed Public Function ") Ċ Ċ package Ġio . etcd . recipes . barrier Ċ Ċ import Ġcom . github . p amb rose . common . delegate . Atomic Deleg ates . atomic Boolean Ċ import Ġcom . github . p amb rose . common . delegate . Atomic Deleg ates . nullable Reference Ċ import Ġcom . github . p amb rose . common . time . time Unit To Duration Ċ import Ġcom . github . p amb rose . common . util . random Id Ċ import Ġio . etcd . jet cd . Client Ċ import Ġio . etcd . jet cd . support . Closeable Client Ċ import Ġio . etcd . jet cd . watch . Watch Event . EventType . DELETE Ċ import Ġio . etcd . recipes . barrier . Distributed Double Barrier . Companion . default ClientId Ċ import Ġio . etcd . recipes . common .* Ċ import Ġmu . K Logging Ċ import Ġjava . util . concurrent . Count DownLatch Ċ import Ġjava . util . concurrent . TimeUnit Ċ import Ġkotlin . time . Duration Ċ Ċ @ Jvm Over loads Ċ fun Ġ< T > Ġwith Distributed Barrier ( ĊĠ Ġclient : ĠClient , ĊĠ Ġbarrier Path : ĠString , ĊĠ Ġlease T tl Secs : ĠLong Ġ= ĠEt cd Connector . default T tl Secs , ĊĠ Ġwait On Missing Bar riers : ĠBoolean Ġ= Ġtrue , ĊĠ ĠclientId : ĠString Ġ= Ġdefault ClientId (), ĊĠ Ġreceiver : ĠDistributed Barrier . () Ġ-> ĠT Ċ ): ĠT Ġ= ĊĠ ĠDistributed Barrier ( client , Ġbarrier Path , Ġlease T tl Secs , Ġwait On Missing Bar riers , ĠclientId ). use Ġ{ Ġit . receiver () Ġ} Ċ Ċ class ĠDistributed Barrier Ċ @ Jvm Over loads Ċ constructor ( ĊĠ Ġclient : ĠClient , ĊĠ Ġval Ġbarrier Path : ĠString , ĊĠ Ġval Ġlease T tl Secs : ĠLong Ġ= Ġdefault T tl Secs , ĊĠ Ġprivate Ġval Ġwait On Missing Bar riers : ĠBoolean Ġ= Ġtrue , ĊĠ Ġval ĠclientId : ĠString Ġ= Ġdefault ClientId () Ċ ) Ġ: ĠEt cd Connector ( client ) Ġ{ ĊĊĠ Ġprivate Ġvar Ġkeep Alive Lease Ġby Ġnullable Reference < Closeable Client ? >( null ) ĊĠ Ġprivate Ġvar Ġbarrier Removed Ġby Ġatomic Boolean ( false ) ĊĊĠ Ġinit Ġ{ ĊĠĠĠ Ġrequire ( barrier Path . isNotEmpty ()) Ġ{ Ġ" Barrier Ġpath Ġcannot Ġbe Ġempty " Ġ} ĊĠ Ġ} ĊĊĠ Ġfun Ġis Barrier Set (): ĠBoolean Ġ{ ĊĠĠĠ Ġcheck Close Not Called () ĊĠĠĠ Ġreturn Ġclient . is Key Present ( barrier Path ) ĊĠ Ġ} ĊĊĠ Ġ@ Synchron ized ĊĠ Ġfun Ġset Barrier (): ĠBoolean Ġ{ ĊĠĠĠ Ġcheck Close Not Called () ĊĠĠĠ Ġreturn Ġif Ġ( client . is Key Present ( barrier Path )) ĊĠĠĠĠĠ Ġfalse ĊĠĠĠ Ġelse Ġ{ ĊĠĠĠĠĠ Ġ// ĠCreate Ġunique Ġtoken Ġto Ġavoid Ġcollision Ġfrom Ġclients Ġwith Ġsame Ġid ĊĠĠĠĠĠ Ġval Ġunique Token Ġ= Ġ"$ clientId :${ random Id ( token Length )}" ĊĊĠĠĠĠĠ Ġ// ĠPrime Ġlease Ġwith Ġ 2 Ġseconds Ġto Ġgive Ġkeep Alive Ġa Ġchance Ġto Ġget Ġstarted ĊĠĠĠĠĠ Ġval Ġlease Ġ= Ġclient . lease Grant ( Duration . seconds ( lease T tl Secs )) ĊĊĠĠĠĠĠ Ġ// ĠDo Ġa ĠC AS Ġon Ġthe Ġkey Ġname . ĠIf Ġit Ġis Ġnot Ġfound , Ġthen Ġset Ġit ĊĠĠĠĠĠ Ġval Ġtxn Ġ= ĊĠĠĠĠĠĠĠ Ġclient . transaction Ġ{ ĊĠĠĠĠĠĠĠĠĠ ĠIf ( barrier Path . does NotExist ) ĊĠĠĠĠĠĠĠĠĠ ĠThen ( barrier Path . set To ( unique Token , Ġput Option Ġ{ Ġwith Lease Id ( lease . id ) Ġ})) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠ Ġ// ĠCheck Ġto Ġsee Ġif Ġunique Ġvalue Ġwas Ġsuccessfully Ġset Ġin Ġthe ĠC AS Ġstep ĊĠĠĠĠĠ Ġif Ġ( txn . is Succeeded Ġ&& Ġclient . getValue ( barrier Path )?. asString Ġ== Ġunique Token ) Ġ{ ĊĠĠĠĠĠĠĠ Ġkeep Alive Lease Ġ= Ġclient . keep Alive ( lease ) ĊĠĠĠĠĠĠĠ Ġtrue ĊĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠ Ġfalse ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ@ Synchron ized ĊĠ Ġfun Ġremove Barrier (): ĠBoolean Ġ{ ĊĠĠĠ Ġcheck Close Not Called () ĊĠĠĠ Ġreturn Ġif Ġ( barrier Removed ) Ġ{ ĊĠĠĠĠĠ Ġfalse ĊĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠ Ġkeep Alive Lease ?. close () ĊĠĠĠĠĠ Ġkeep Alive Lease Ġ= Ġnull ĊĊĠĠĠĠĠ Ġclient . delete Key ( barrier Path ) ĊĊĠĠĠĠĠ Ġbarrier Removed Ġ= Ġtrue ĊĊĠĠĠĠĠ Ġtrue ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ@ Throws ( InterruptedException :: class ) ĊĠ Ġfun Ġwait On Barrier (): ĠBoolean Ġ= Ġwait On Barrier ( Duration . days ( Long . MAX _ VALUE )) ĊĊĠ Ġ@ Throws ( InterruptedException :: class ) ĊĠ Ġfun Ġwait On Barrier ( timeout : ĠLong , Ġtime Unit : ĠTimeUnit ): ĠBoolean Ġ= ĊĠĠĠ Ġwait On Barrier ( time Unit To Duration ( timeout , Ġtime Unit )) ĊĊĠ Ġ@ Throws ( InterruptedException :: class ) ĊĠ Ġfun Ġwait On Barrier ( timeout : ĠDuration ): ĠBoolean Ġ{ ĊĠĠĠ Ġcheck Close Not Called () ĊĊĠĠĠ Ġ// ĠCheck Ġif Ġbarrier Ġis Ġpresent Ġbefore Ġusing Ġwatcher ĊĠĠĠ Ġreturn Ġif Ġ(! wait On Missing Bar riers Ġ&& Ġ! is Barrier Set ()) ĊĠĠĠĠĠ Ġtrue ĊĠĠĠ Ġelse Ġ{ ĊĠĠĠĠĠ Ġval Ġwait Latch Ġ= ĠCount DownLatch ( 1 ) ĊĠĠĠĠĠ Ġval Ġwatch Option Ġ= Ġwatch Option Ġ{ Ġwith No Put ( true ) Ġ} ĊĊĠĠĠĠĠ Ġclient . with Watcher ( barrier Path , ĊĠĠĠĠĠĠĠ Ġwatch Option , ĊĠĠĠĠĠĠĠ Ġ{ Ġwatch Response Ġ-> ĊĠĠĠĠĠĠĠĠĠ Ġfor Ġ( event Ġin Ġwatch Response . events ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( event . eventType Ġ== ĠDELETE ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwait Latch . count Down () ĊĠĠĠĠĠĠĠ Ġ}) Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠCheck Ġone Ġmore Ġtime Ġin Ġcase Ġwatch Ġmissed Ġthe Ġdelete Ġjust Ġafter Ġlast Ġcheck ĊĠĠĠĠĠĠĠ Ġif Ġ(! wait On Missing Bar riers Ġ&& Ġ! is Barrier Set ()) ĊĠĠĠĠĠĠĠĠĠ Ġwait Latch . count Down () ĊĊĠĠĠĠĠĠĠ Ġwait Latch . await ( timeout . in Whole Milliseconds , ĠTimeUnit . MILLISECONDS ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ@ Synchron ized ĊĠ Ġoverride Ġfun Ġclose () Ġ{ ĊĠĠĠ Ġif Ġ( close Called ) ĊĠĠĠĠĠ Ġreturn ĊĊĠĠĠ Ġkeep Alive Lease ?. close () ĊĠĠĠ Ġkeep Alive Lease Ġ= Ġnull ĊĊĠĠĠ Ġsuper . close () ĊĠ Ġ} ĊĊĠ Ġcompanion Ġobject Ġ: ĠK Logging () Ġ{ ĊĠĠĠ Ġinternal Ġfun Ġdefault ClientId () Ġ= Ġ"${ Distributed Barrier :: class . simple Name }:${ random Id ( token Length )}" ĊĠ Ġ} Ċ }