package Ġlearning . zookeeper Ċ Ċ import Ġorg . apache . zookeeper .* Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġjava . util .* Ċ import Ġjava . util . concurrent . TimeUnit Ċ Ċ class ĠMaster Async ( val Ġhost Port : ĠString ) Ġ: ĠWatch er Ġ{ ĊĊĠĠĠ Ġprivate Ġval ĠLOG Ġ= ĠLoggerFactory . getLogger ( this . java Class ) ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġzk : ĠZ oo Keeper ĊĊĠĠĠ Ġprivate Ġval Ġserver Id Ġ= ĠInteger . to HexString ( Random (). nextInt ()) ĊĠĠĠ Ġvar Ġis Leader Ġ= Ġfalse ĊĊĊĠĠĠ Ġfun Ġstart ZK () Ġ{ ĊĠĠĠĠĠĠĠ Ġzk Ġ= ĠZ oo Keeper ( host Port , Ġ 1 5 0 0 0 , Ġthis ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġmaster Check Callback ( rc : ĠInt ) Ġ{ ĊĠĠĠĠĠĠĠ Ġwhen Ġ( Keeper Exception . Code . get ( rc )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠK eeper Exception . Code . CONNECTION LO SS Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheck Master () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠK eeper Exception . Code . NON ODE Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun For Master () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġfun Ġcheck Master () Ġ{ ĊĠĠĠĠĠĠĠ Ġzk . getData ("/ master Async ", Ġfalse , Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrc , Ġpath , Ġctx , Ġdata , Ġstat Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmaster Check Callback ( rc ) ĊĠĠĠĠĠĠĠ Ġ}, Ġnull ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġmaster Create Callback ( rc : ĠInt ) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġwhen Ġ( Keeper Exception . Code . get ( rc )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠK eeper Exception . Code . CONNECTION LO SS Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheck Master () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠK eeper Exception . Code . OK Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis Leader Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis Leader Ġ= Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġprintln (" I 'm Ġ${ if Ġ( is Leader ) Ġ"" Ġelse Ġ" not "} Ġthe Ġleader ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġrun For Master () Ġ{ ĊĠĠĠĠĠĠĠ Ġzk . create ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ"/ master ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġserver Id . toByteArray (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠZ oo Defs . Ids . OPEN _ ACL _ UN SAFE , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCreate Mode . EP HE MER AL , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ{ Ġrc , Ġpath , Ġctx , Ġname Ġ-> Ġmaster Create Callback ( rc ) Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġstop ZK () Ġ{ ĊĠĠĠĠĠĠĠ Ġzk . close () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġprocess ( event : ĠW atched Event ) Ġ{ ĊĠĠĠĠĠĠĠ Ġprintln ( event ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġbootstrap () Ġ{ ĊĠĠĠĠĠĠĠ Ġcreate Parent ("/ workers ", ĠByteArray ( 0 )) ĊĠĠĠĠĠĠĠ Ġcreate Parent ("/ assign ", ĠByteArray ( 0 )) ĊĠĠĠĠĠĠĠ Ġcreate Parent ("/ tasks ", ĠByteArray ( 0 )) ĊĠĠĠĠĠĠĠ Ġcreate Parent ("/ status ", ĠByteArray ( 0 )) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġcreate Parent ( path : ĠString , Ġdata : ĠByteArray ) Ġ{ ĊĠĠĠĠĠĠĠ Ġzk . create ( path , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdata , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠZ oo Defs . Ids . OPEN _ ACL _ UN SAFE , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCreate Mode . PER SIST ENT , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ{ Ġrc , Ġpath , Ġctx , Ġname Ġ-> Ġcreate Parent Callback ( rc , Ġpath , Ġctx ) Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdata ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġcreate Parent Callback ( rc : ĠInt , Ġpath : ĠString , Ġctx : ĠAny ) Ġ{ ĊĠĠĠĠĠĠĠ Ġwhen Ġ( Keeper Exception . Code . get ( rc )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠK eeper Exception . Code . CONNECTION LO SS Ġ-> Ġcreate Parent ( path , Ġctx Ġas ĠByteArray ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠK eeper Exception . Code . OK Ġ-> ĠLOG . info (" Parent Ġcreated ") ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠK eeper Exception . Code . NODE EXISTS Ġ-> ĠLOG . warn (" Parent Ġalready Ġregistered : Ġ$ path ") ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠLOG . error (" Something Ġwent Ġwrong : Ġ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠK eeper Exception . create ( Keeper Exception . Code . get ( rc ), Ġpath )) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ fun Ġmain ( args : ĠArray < String >) Ġ{ ĊĠĠĠ Ġval Ġmaster Ġ= ĠMaster Async (" 1 2 7 . 0 . 0 . 1 : 2 1 8 1 ") ĊĊĠĠĠ Ġmaster . start ZK () ĊĠĠĠ Ġmaster . bootstrap () ĊĠĠĠ Ġmaster . run For Master () ĊĊĠĠĠ Ġif Ġ( master . is Leader ) Ġ{ ĊĠĠĠĠĠĠĠ Ġprintln (" I 'm Ġthe Ġleader ") ĊĠĠĠĠĠĠĠ ĠTimeUnit . SECONDS . sleep ( 6 0 ) ĊĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠ Ġprintln (" Some one Ġelse Ġis Ġthe Ġleader ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġmaster . stop ZK () Ċ } Ċ