package Ġorg . ker ala . core . consensus . replication Ċ Ċ import Ġkotlinx . coroutines . delay Ċ import Ġorg . ker ala . core . consensus . Raft Ċ import Ġorg . ker ala . core . consensus . messages . Append Entries Request Ċ import Ġorg . ker ala . core . consensus . messages . Append Entries Response Ċ import Ġorg . ker ala . core . consensus . states . leader . Leader Context Ċ import Ġorg . ker ala . core . runtime . topic . Topic Ċ import Ġorg . ker ala . core . server . cluster . Node Ċ import Ġorg . ker ala . shared . logger Ċ import Ġkotlin . math . max Ċ import Ġkotlin . system . measure Time Millis Ċ Ċ /** Ċ Ġ* ĠNode Replication Controller Ġreplic ates Ġthe Ġgiven Ġra ft Ġstate Ġto Ġthe Ġtarget Ġ{$ link ĠNode }. Ċ Ġ* Ċ Ġ* Ġ@ see ĠNode Ċ Ġ* Ġ@ see ĠRep lic ator Ċ Ġ*/ Ċ class ĠNode Replication Controller ( ĊĠĠĠ Ġval Ġra ft : ĠR aft , ĊĠĠĠ Ġval Ġtopic : ĠTopic , ĊĠĠĠ Ġval Ġtarget : ĠNode , ĊĠĠĠ Ġval Ġleader Context : ĠLeader Context , ĊĠĠĠ Ġprivate Ġval Ġbroadcast Interval : ĠLong , ĊĠĠĠ Ġprivate Ġval Ġreplic ator Strategy : ĠRep lic ator Strategy Ġ= ĠRep lic ator Strategy ( topic , Ġra ft ) Ċ ) Ġ{ ĊĠ Ġinit Ġ{ ĊĠĠĠ Ġlogger (" replic ating Ġ$ topic Ġ-> Ġ$ target ") ĊĠ Ġ} ĊĊĠ Ġsuspend Ġfun Ġstart () Ġ{ ĊĠĠĠ Ġ/* Ġsend Ġheartbeat Ġto Ġfollow ers Ġas Ġnewly Ġelect ed Ġleader , Ġwe Ġwant ĊĠĠĠĠĠ Ġto Ġensure Ġwe Ġaccount Ġfor Ġoverhead Ġof Ġinitializing Ġthe Ġnew Ġleader Ġstate Ġ*/ ĊĠĠĠ Ġsend Heartbeat () ĊĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠ Ġ/* ĠDelay Ġfor Ġremainder Ġof Ġbroadcast Ġinterval Ġand Ġthe Ġtime Ġtaken Ġto Ġreplicate Ġto Ġtarget . ĠIf Ġexceeded , ĊĠĠĠĠĠĠĠ Ġzero Ġdelay , Ġin Ġwhich Ġcase Ġa Ġfollow er Ġhas Ġprobably Ġtimed Ġout Ġand Ġtransition ed Ġto Ġa Ġcandidate Ġstate . Ġ*/ ĊĠĠĠĠĠ Ġdelay ( max ( broadcast Interval Ġ- Ġmeasure Time Millis Ġ{ Ġreplicate () Ġ}, Ġ 0 )) ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġprivate Ġsuspend Ġfun Ġreplicate () Ġ{ ĊĠĠĠ Ġ/* Ġdetermine Ġnext Ġentries Ġto Ġreplicate Ġ*/ ĊĠĠĠ Ġval Ġnext Index Ġ= Ġleader Context . getNext Index ( target ) ĊĠĠĠ Ġval Ġrequest Ġ= Ġreplic ator Strategy . generate Request ( next Index ) ĊĠĠĠ Ġra ft . cluster Messenger . dispatch Append Entries ( target , Ġrequest , ĠonSuccess Ġ= Ġ{ Ġresponse Ġ-> ĊĠĠĠĠĠ Ġhandle Response ( request , Ġresponse ) ĊĠĠĠ Ġ}) ĊĠ Ġ} ĊĊĠ Ġprivate Ġfun Ġhandle Response ( request : ĠAppend Entries Request , Ġresponse : ĠAppend Entries Response ) Ġ{ ĊĠĠĠ Ġif Ġ( response . is Successful ) Ġ{ ĊĠĠĠĠĠ Ġwith ( leader Context ) Ġ{ ĊĠĠĠĠĠĠĠ Ġupdate Match Index ( target , Ġrequest . prev Log Index Ġ+ Ġrequest . entries . size ) ĊĠĠĠĠĠĠĠ Ġupdate Next Index ( target , Ġrequest . prev Log Index Ġ+ Ġrequest . entries . size Ġ+ Ġ 1 ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠ Ġleader Context . update Next Index ( target , Ġresponse . prev Log Index Ġ+ Ġ 1 ) ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġprivate Ġsuspend Ġfun Ġsend Heartbeat () Ġ{ ĊĠĠĠ Ġwith ( topic . log Facade ) Ġ{ ĊĠĠĠĠĠ Ġval Ġmessage Ġ= ĠAppend Entries Request ( ĊĠĠĠĠĠĠĠĠĠ Ġterm Ġ= Ġra ft . raft Context . current Term , ĊĠĠĠĠĠĠĠĠĠ Ġtopic Id Ġ= Ġtopic . id , ĊĠĠĠĠĠĠĠĠĠ Ġprev Log Term Ġ= Ġlog . last Entry ?. term Ġ?: Ġ 0 , ĊĠĠĠĠĠĠĠĠĠ Ġprev Log Index Ġ= Ġlog . last Index , ĊĠĠĠĠĠĠĠĠĠ Ġleader Id Ġ= Ġra ft . cluster Set . self Node . id , ĊĠĠĠĠĠĠĠĠĠ Ġleader Commit Ġ= Ġlog . commit Index ) ĊĠĠĠĠĠ Ġra ft . cluster Messenger . dispatch Append Entries ( target , Ġmessage ) ĊĠĠĠ Ġ} ĊĠ Ġ} Ċ } Ċ