-
Notifications
You must be signed in to change notification settings - Fork 7
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #43 from parapet-io/iss31
cluster
- Loading branch information
Showing
53 changed files
with
1,869 additions
and
302 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
150 changes: 150 additions & 0 deletions
150
cluster-api/src/main/scala/io/parapet/cluster/api/ClusterApi.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,150 @@ | ||
| package io.parapet.cluster.api | ||
|
|
||
| import io.parapet.core.{Encoder, Event} | ||
|
|
||
| import java.nio.ByteBuffer | ||
|
|
||
| object ClusterApi { | ||
|
|
||
|
|
||
| // @formatter:off | ||
| sealed trait API extends Event | ||
|
|
||
| // --------------------- JOIN ------------------------------------------ // | ||
| object JoinResultCodes { | ||
| val OK = 0 | ||
| val ERROR = 1 // must to be clarified | ||
| } | ||
|
|
||
| // FORMAT | ||
| val TAG_SIZE = 4 | ||
| val NODE_ID_SIZE = 4 | ||
| val ADDR_SIZE = 4 | ||
| val GROUP_SIZE = 4 | ||
| val RESULT_CODE_SIZE = 4 | ||
|
|
||
| case class Join(nodeId: String, address: String, group: String) extends API | ||
| case class JoinResult(nodeId: String, code: Int) extends API | ||
|
|
||
| // ---------------------- Result ---------------------------------------- // | ||
| case class Result(code: Int, msg: String) extends API | ||
|
|
||
| object ResultCodes { | ||
| val OK = 0 | ||
| val ERROR = 1 | ||
| } | ||
| // FORMAT | ||
| val CODE_SIZE = 4 | ||
| val MSG_SIZE = 4 | ||
|
|
||
| // ---------------------- NodeInfo ---------------------------------------- // | ||
| case class GetNodeInfo(senderId: String, id: String) extends API | ||
| case class NodeInfo(address: String, code: Int) extends API | ||
| object NodeInfoCodes { | ||
| val OK = 0 | ||
| val NODE_NOT_FOUND = 1 | ||
| val ERROR = 2 | ||
| } | ||
|
|
||
| // TAGS | ||
| val JOIN_TAG = 20 | ||
| val JOIN_RESULT_TAG = 21 | ||
| val RESULT_TAG = 22 | ||
| val GET_NODE_INFO_TAG = 23 | ||
| val NODE_INFO = 24 | ||
|
|
||
| // @formatter:on | ||
|
|
||
| val encoder: Encoder = new Encoder { | ||
| override def write(e: Event): Array[Byte] = { | ||
| e match { | ||
| case Join(nodeId, address, group) => | ||
| val nodeIdBytes = nodeId.getBytes() | ||
| val addressBytes = address.getBytes() | ||
| val groupBytes = group.getBytes() | ||
| val buf = ByteBuffer.allocate(TAG_SIZE + | ||
| (NODE_ID_SIZE + nodeIdBytes.length) + | ||
| (GROUP_SIZE + groupBytes.length) + | ||
| (ADDR_SIZE + addressBytes.length)) | ||
| // write | ||
| buf.putInt(JOIN_TAG) | ||
| putWithSize(buf, nodeIdBytes) | ||
| putWithSize(buf, addressBytes) | ||
| putWithSize(buf, groupBytes) | ||
| buf.rewind() | ||
| buf.array() | ||
| case JoinResult(nodeId, code) => | ||
| val nodeIdBytes = nodeId.getBytes() | ||
| val buf = ByteBuffer.allocate(TAG_SIZE + (NODE_ID_SIZE + nodeIdBytes.length) + RESULT_CODE_SIZE) | ||
| buf.putInt(JOIN_RESULT_TAG) | ||
| putWithSize(buf, nodeIdBytes) | ||
| buf.putInt(code) | ||
| buf.rewind() | ||
| buf.array() | ||
| case Result(code, msg) => | ||
| val msgBytes = msg.getBytes() | ||
| val buf = ByteBuffer.allocate(TAG_SIZE + CODE_SIZE + (MSG_SIZE + msgBytes.length)) | ||
| buf.putInt(RESULT_TAG) | ||
| buf.putInt(code) | ||
| putWithSize(buf, msgBytes) | ||
| buf.rewind() | ||
| buf.array() | ||
| case GetNodeInfo(senderId, id) => | ||
| val senderIdBytes = senderId.getBytes() | ||
| val idBytes = id.getBytes() | ||
| val buf = ByteBuffer.allocate(4 + 4 + senderIdBytes.length + 4 + idBytes.length) | ||
| buf.putInt(GET_NODE_INFO_TAG) | ||
| putWithSize(buf, senderIdBytes) | ||
| putWithSize(buf, idBytes) | ||
| buf.rewind() | ||
| buf.array() | ||
| case NodeInfo(address, code) => | ||
| val addressBytes = address.getBytes | ||
| val buf = ByteBuffer.allocate(TAG_SIZE + 4 + CODE_SIZE + addressBytes.length) | ||
| .putInt(NODE_INFO) | ||
| .putInt(addressBytes.length) | ||
| .put(addressBytes) | ||
| .putInt(code) | ||
| buf.rewind() | ||
| buf.array() | ||
| case _ => throw new UnsupportedOperationException() | ||
| } | ||
| } | ||
|
|
||
| override def read(data: Array[Byte]): Event = { | ||
| val buf = ByteBuffer.wrap(data) | ||
| val tag = buf.getInt | ||
| tag match { | ||
| case JOIN_TAG => | ||
| val nodeId = getString(buf) | ||
| val address = getString(buf) | ||
| val group = getString(buf) | ||
| Join(nodeId = nodeId, address = address, group = group) | ||
| case JOIN_RESULT_TAG => | ||
| val nodeId = getString(buf) | ||
| val code = buf.getInt | ||
| JoinResult(nodeId, code) | ||
| case RESULT_TAG => | ||
| val code = buf.getInt | ||
| val msg = getString(buf) | ||
| Result(code, msg) | ||
| case GET_NODE_INFO_TAG => GetNodeInfo(getString(buf), getString(buf)) | ||
| case NODE_INFO => NodeInfo(getString(buf), buf.getInt) | ||
| case _ => throw new UnsupportedOperationException() | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private def putWithSize(buf: ByteBuffer, data: Array[Byte]): Unit = { | ||
| buf.putInt(data.length) | ||
| buf.put(data) | ||
| } | ||
|
|
||
| private def getString(buf: ByteBuffer): String = { | ||
| val len = buf.getInt() | ||
| val data = new Array[Byte](len) | ||
| buf.get(data) | ||
| new String(data) | ||
| } | ||
|
|
||
| } |
15 changes: 15 additions & 0 deletions
15
cluster-api/src/test/scala/io/parapet/cluster/api/EncoderSpec.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,15 @@ | ||
| package io.parapet.cluster.api | ||
|
|
||
| import io.parapet.cluster.api.ClusterApi._ | ||
| import org.scalatest.funsuite.AnyFunSuite | ||
| import org.scalatest.matchers.should.Matchers._ | ||
|
|
||
| class EncoderSpec extends AnyFunSuite { | ||
|
|
||
| test("join") { | ||
| val join = Join("1", "localhost:8080", "2") | ||
| val data = encoder.write(join) | ||
| encoder.read(data) shouldBe join | ||
| } | ||
|
|
||
| } |
71 changes: 71 additions & 0 deletions
71
cluster-node/src/main/scala/io/parapet/cluster/node/Interface.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| package io.parapet.cluster.node | ||
|
|
||
| import scala.util.Try | ||
|
|
||
| trait Interface { | ||
|
|
||
| /** Connects to the leader. | ||
| */ | ||
| def connect(): Unit | ||
|
|
||
| /** Joins a cluster. | ||
| * @param group the node group | ||
| * @return result | ||
| */ | ||
| def join(group: String): Try[Unit] | ||
|
|
||
| /** Leaves the given node group. | ||
| * @param group the node group | ||
| * @return result | ||
| */ | ||
| def leave(group: String): Try[Unit] | ||
|
|
||
| /** Sends a request. | ||
| * | ||
| * @param req the request | ||
| * @return result | ||
| */ | ||
| def send(req: Req): Try[Unit] | ||
|
|
||
| /** | ||
| * Send a request and waits for response. | ||
| * Use for a strict REQ-REP dialog. | ||
| * | ||
| * @param req request | ||
| * @param handler message handler | ||
| * @return result | ||
| */ | ||
| def send(req: Req, handler: Array[Byte] => Unit): Try[Unit] | ||
|
|
||
| /** | ||
| * Sends a reply. | ||
| * Use for a strict REQ-REP dialog. | ||
| * | ||
| * @param rep the reply | ||
| * @return result | ||
| */ | ||
| def send(rep: Rep): Try[Unit] | ||
|
|
||
| /** Sends a message to all nodes in the group. | ||
| * | ||
| * @param group the node group | ||
| * @param data the data to send | ||
| * @return result | ||
| */ | ||
| def broadcast(group: String, data: Array[Byte]): Try[Unit] | ||
|
|
||
| /** Gets current leader. | ||
| * | ||
| * @return leader | ||
| */ | ||
| def leader: Option[String] | ||
|
|
||
| /** Gets all registered nodes. | ||
| * @return registered nodes | ||
| */ | ||
| def getNodes: Try[Seq[String]] | ||
|
|
||
| /** Disconnects from the cluster and closes open network connections. | ||
| */ | ||
| def close(): Unit | ||
| } |
7 changes: 7 additions & 0 deletions
7
cluster-node/src/main/scala/io/parapet/cluster/node/MessageHandler.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,7 @@ | ||
| package io.parapet.cluster.node | ||
|
|
||
| trait MessageHandler { | ||
|
|
||
| def handle(req: Req): Unit | ||
|
|
||
| } |
Oops, something went wrong.