libraryDependencies += "io.ckite" %% "ckite-core" % "0.2.1"

libraryDependencies += "io.ckite" %% "ckite-finagle" % "0.2.1"

libraryDependencies += "io.ckite" %% "ckite-mapdb" % "0.2.1"

//KVStore is an in-memory distributed Map allowing Puts and Gets operations
class KVStore extends StateMachine {

  private var map = Map[String, String]()
  private var lastIndex: Long = 0

  //Called when a consensus has been reached for a WriteCommand
  //index associated to the write is provided to implement your own persistent semantics
  //see lastAppliedIndex
  def applyWrite = {
    case (index, Put(key: String, value: String)) => {
      map.put(key, value)
      lastIndex = index
      value
    }
  }

  //called when a read command has been received
  def applyRead = {
    case Get(key) => map.get(key)
  }

  //CKite needs to know the last applied write on log replay to 
  //provide exactly-once semantics
  //If no persistence is needed then state machines can just return zero
  def getLastAppliedIndex: Long = lastIndex

  //called during Log replay on startup and upon installSnapshot requests
  def restoreSnapshot(byteBuffer: ByteBuffer) = {
    map = Serializer.deserialize[Map[String, String]](byteBuffer.array())
  }
  //called when Log compaction is required
  def takeSnapshot(): ByteBuffer = ByteBuffer.wrap(Serializer.serialize(map))

}

//WriteCommands are replicated under Raft rules
case class Put(key: String, value: String) extends WriteCommand[String]

//ReadCommands are not replicated but forwarded to the Leader
case class Get(key: String) extends ReadCommand[Option[String]]

val ckite = CKiteBuilder().listenAddress("node1:9091").rpc(FinagleThriftRpc) //Finagle based transport
                          .stateMachine(new KVStore()) //KVStore is an implementation of the StateMachine trait
                          .bootstrap(true) //bootstraps a new cluster. only needed just the first time for the very first node
                          .build

val ckite = CKiteBuilder().listenAddress("localhost:9091").rpc(FinagleThriftRpc)
                          .members(Seq("localhost:9092","localhost:9093")) //optional seeds to join the cluster
                          .minElectionTimeout(1000).maxElectionTimeout(1500) //optional
                          .heartbeatsPeriod(250) //optional. period to send heartbeats interval when being Leader
                          .dataDir("/home/ckite/data") //dataDir for persistent state (log, terms, snapshots, etc...)
                          .stateMachine(new KVStore()) //KVStore is an implementation of the StateMachine trait
                          .sync(false) //disables log sync to disk
                          .flushSize(10) //max batch size when flushing log to disk
                          .build

ckite.start()

//this Put command is forwarded to the Leader and applied under Raft rules
val writeFuture:Future[String] = ckite.write(Put("key1","value1")) 

//consistent read commands are forwarded to the Leader
val readFuture:Future[Option[String]] = ckite.read(Get("key1")) 

//as write commands, cluster membership changes are forwarded to the Leader
ckite.addMember("someHost:9094")

//as write commands, cluster membership changes are forwarded to the Leader
ckite.removeMember("someHost:9094")

//alternatively you can read from its local state machine allowing possible stale values
val value = ckite.readLocal(Get("key1")) 

//if necessary waits for elections to end
ckite.isLeader() 

ckite.stop()

val ckite = CKiteBuilder().listenAddress("node1:9091").rpc(FinagleThriftRpc)
                          .dataDir("/home/ckite/data") //dataDir for persistent state (log, terms, snapshots, etc...)
                          .stateMachine(new KVStore()) //KVStore is an implementation of the StateMachine trait
                          .bootstrap(true) //bootstraps a new cluster. only needed just the first time for the very first node
                          .build