refactor: merge exchange service

rpc/src/main/protobuf/worker.proto

@@ -9,6 +9,7 @@ service Worker {
   rpc Sort (SortRequest) returns (SortReply) {}
   rpc Partition (PartitionRequest) returns (PartitionReply) {}
   rpc Exchange (ExchangeRequest) returns (ExchangeReply) {}
+  rpc SaveBlock (SaveBlockRequest) returns (SaveBlockReply) {}
   rpc Merge (MergeRequest) returns (MergeReply) {}
 }
 
@@ -36,6 +37,12 @@ message ExchangeRequest {
 
 message ExchangeReply {}
 
+message SaveBlockRequest {
+  bytes block = 1;
+}
+
+message SaveBlockReply {}
+
 message MergeRequest {}
 
 message MergeReply {}

rpc/src/main/scala/WorkerClient.scala

@@ -3,13 +3,11 @@ package kr.ac.postech.paranode.rpc
 import com.google.protobuf.ByteString
 import io.grpc.ManagedChannel
 import io.grpc.ManagedChannelBuilder
-import kr.ac.postech.paranode.core.KeyRange
-import kr.ac.postech.paranode.core.WorkerMetadata
+import kr.ac.postech.paranode.core.{Block, KeyRange, WorkerMetadata}
 
 import java.util.concurrent.TimeUnit
 import java.util.logging.Logger
 import scala.concurrent.Future
-
 import worker._
 import worker.WorkerGrpc.WorkerStub
 import common.{
@@ -89,6 +87,16 @@ class WorkerClient private (
     stub.exchange(request)
   }
 
+  def saveBlock(
+      block: Block
+  ): Future[SaveBlockReply] = {
+    val request = SaveBlockRequest(
+      ByteString.copyFrom(block.toChars.map(_.toByte).toArray)
+    )
+
+    stub.saveBlock(request)
+  }
+
   def merge(): Future[MergeReply] = {
     val request = MergeRequest()
     stub.merge(request)

rpc/src/main/scala/WorkerServer.scala

@@ -12,9 +12,9 @@ import scala.concurrent.Future
 import scala.concurrent.Promise
 import scala.reflect.io.Directory
 import scala.reflect.io.Path
-
 import common.{WorkerMetadata => RpcWorkerMetadata}
 import worker._
+import java.util.UUID
 
 class WorkerServer(
     executionContext: ExecutionContext,
@@ -165,6 +165,32 @@ class WorkerServer(
       Future.sequence(futures).map(_ => new ExchangeReply())
     }
 
+    override def saveBlock(
+        request: SaveBlockRequest
+    ): Future[SaveBlockReply] = {
+      val promise = Promise[SaveBlockReply]
+
+      Future {
+        logger.debug(s"[WorkerServer] SaveBlock ($request)")
+
+        val block = Block.fromBytes(
+          LazyList.from(request.block.toByteArray)
+        )
+
+        val path = outputDirectory / UUID.randomUUID().toString
+
+        logger.debug(s"[WorkerServer] Writing block to $path")
+
+        block.writeTo(path)
+
+        logger.debug(s"[WorkerServer] Wrote block to $path")
+
+        promise.success(new SaveBlockReply())
+      }
+
+      promise.future
+    }
+
     override def merge(request: MergeRequest): Future[MergeReply] = {
       val promise = Promise[MergeReply]