Building my own blockchain
Table of Contents
Building the Basic Block
A blockchain is made up of fundamental units called blocks. Each block contains key information including a list of transactions, the hash of the previous block, the block's index in the chain, a timestamp, and its own hash. To begin, we define the case classes required to instantiate a block.
1case class Transaction(sender: String, receiver: String, amount: Int)
2
3case class Block(index: Int, hash: String, prevHash: String, proof: Long, timestamp: Long, transactions: List[Transaction])Next, we implement the proof-of-work mechanism. Each block must be validated by computing a proof. The hash of the block is calculated using the block's index, the previous hash, the list of transactions, and the proof value. We use SHA-256 as the hashing algorithm. Mining a block involves finding a proof such that the resulting hash starts with a given number of zeros (determined by the difficulty level). This is done recursively until a valid proof is found.
1def computeHash(index: Int, prevHash: String, transactions: List[Transaction], proof: Long): String = {
2 val transactionsStr = transactions.map(transaction =>
3 s"${transaction.sender} -> ${transaction.receiver} amount: ${transaction.amount}"
4 ).mkString("|")
5 val hashStr = s"$index$prevHash$transactionsStr$proof"
6 val bytes = hashStr.getBytes
7 val hash = MessageDigest.getInstance("SHA-256").digest(bytes)
8 hash.map("%02x".format(_)).mkString
9}
10
11def mineBlock(index: Int, prevHash: String, transactions: List[Transaction]) = {
12 val timeStamp = System.currentTimeMillis()
13 val prefix = "0" * difficulty
14
15 @tailrec
16 def findProof(proof: Int): (String, Long) = {
17 val hash = computeHash(index, prevHash, transactions, proof)
18 if (hash.startsWith(prefix)) (hash, proof)
19 else findProof(proof + 1)
20 }
21
22 val (hash, proof) = findProof(0)
23 (hash, proof, timeStamp)
24}Once we have the mineBlock function, we can proceed to create blocks. Every blockchain starts with a genesis block, the first block in the chain. Below, we define methods to create the genesis block and subsequent blocks. Additionally, we implement isValidProof to verify the integrity of a block—particularly useful for validating blocks received through gossip protocols from other nodes.
1def createGenesisBlock(): Block = {
2 val initialTransaction = Transaction("Genesis", "miner", 50)
3 val transactions = List(initialTransaction)
4 val (hash, proof, timestamp) = mineBlock(0, "0", transactions)
5 Block(0, hash, "0", proof, timestamp, transactions)
6}
7
8def createNextBlock(prevBlock: Block, transactions: List[Transaction]) = {
9 val currIndex = prevBlock.index + 1
10 val (hash, proof, timestamp) = mineBlock(currIndex, prevBlock.hash, transactions)
11 Block(currIndex, hash, prevBlock.hash, proof, timestamp, transactions)
12}
13
14// Verifies if the block's hash and proof are valid
15def isValidProof(block: Block): Boolean = {
16 val prefix = "0" * difficulty
17 val hashComputed = computeHash(block.index, block.prevHash, block.transactions, block.proof)
18 hashComputed == block.hash && hashComputed.startsWith(prefix)
19}Building Our Broker
Next, we define our Broker actor. The broker acts as a transaction pool within each node. It manages all unconfirmed transactions and handles the following messages:
- AddTransaction: adds a new transaction to the pool.
- GetTransaction: retrieves the list of current unconfirmed transactions.
- ClearTransactions: clears the transaction pool, typically after a block has been successfully mined.
1sealed trait BrokerEvent
2case class AddTransactionEvent(transaction: Transaction) extends BrokerEvent
3case class GetTransactionEvent(replyTo: ActorRef[List[Transaction]]) extends BrokerEvent
4case object ClearTransactionEvent extends BrokerEventNow we implement the broker actor by defining its behavior in response to each event. When it receives an AddTransactionEvent, the new transaction is appended to the current list. For GetTransactionEvent, it responds with the current list of transactions. And for ClearTransactionEvent, it resets the transaction list to empty.
1// Broker manages the pending transactions within a node
2def apply(): Behavior[BrokerEvent] =
3 brokerBehavior(List.empty)
4
5private def brokerBehavior(
6 pendingTransactions: List[Transaction]
7): Behavior[BrokerEvent] = {
8
9 Behaviors.receive { (context, message) =>
10 message match {
11 case AddTransactionEvent(transaction) =>
12 brokerBehavior(pendingTransactions :+ transaction)
13
14 case GetTransactionEvent(replyTo) =>
15 replyTo ! pendingTransactions
16 Behaviors.same
17
18 case ClearTransactionEvent =>
19 brokerBehavior(List.empty)
20 }
21 }
22}Setting Up the Miner
In this section, we'll set up the miner—the component responsible for validating incoming blocks and mining new ones from the current list of transactions. You can think of the miner as a worker that handles the block generation process.
To begin, we define the messages the miner actor can receive. These include:
To begin, we define the messages the miner actor can receive. These include:
- ValidateBlock: used to verify the integrity and validity of a received block.
- MineCurrentBlockMinerEvent: instructs the miner to start mining the current set of transactions.
- MiningWorkflowEvent messages such as obtainLastHashEvent, obtainLastIndexEvent, and mineBlock which guide the step-by-step mining process.
- MiningFailed: represents any error encountered during mining.
1sealed trait MinerEvent
2case class ValidateBlock(block: Block, replyTo: ActorRef[Boolean]) extends MinerEvent
3
4case object MineCurrentBlockMinerEvent extends MinerEvent
5sealed trait MiningWorkflowEvent extends MinerEvent
6case class obtainLastHashEvent(transactions: List[Transaction]) extends MiningWorkflowEvent
7case class obtainLastIndexEvent(transactions: List[Transaction], prevHash: String) extends MiningWorkflowEvent
8case class mineBlock(transactions: List[Transaction], prevHash: String, index: Int) extends MiningWorkflowEvent
9
10case class MiningFailed(error: String) extends MinerEventWe define a coinbase transaction as a special transaction that rewards the miner for successfully mining a block. This transaction is added to the list of transactions before the block is created.
In the apply method, we initialize the miner actor with references to both the broker and blockchain actors. These references are essential: the miner uses the broker to fetch the current list of pending transactions, and interacts with the blockchain actor to retrieve the previous block's hash and index. This information is required to construct a valid new block.
1def createCoinBaseTransaction(): Transaction = {
2 Transaction("Network", "Miner", 50)
3}
4
5def apply(
6 brokerActor: ActorRef[BrokerDurable.BrokerEvent],
7 blockchainActor: ActorRef[BlockChainDurable.BlockChainEvent]
8): Behavior[MinerEvent] =
9 minerBehavior(brokerActor, blockchainActor)
10Here's the main behavior implementation for the miner actor. It processes the messages, interacts with the broker and blockchain actors, and orchestrates the mining workflow.
1private def minerBehavior(
2 brokerActor: ActorRef[BrokerDurable.BrokerEvent],
3 blockchainActor: ActorRef[BlockChainDurable.BlockChainEvent]
4): Behavior[MinerEvent] = {
5 Behaviors.setup { context =>
6 implicit val timeout: Timeout = 3.seconds
7
8 Behaviors.receive { (context, message) =>
9 message match {
10 case ValidateBlock(block, replyTo) =>
11 val isValid = BlockDurable.isValidProof(block)
12 replyTo ! isValid
13 Behaviors.same
14
15 case MineCurrentBlockMinerEvent =>
16 context.log.info("Received mine block event")
17 context.ask(brokerActor, ref => GetTransactionEvent(ref)) {
18 case Success(transactions) =>
19 val coinbaseTransaction = createCoinBaseTransaction()
20 obtainLastHashEvent(transactions :+ coinbaseTransaction)
21 case Failure(ex) =>
22 MiningFailed(ex.getMessage)
23 }
24 Behaviors.same
25
26 case obtainLastHashEvent(transactions) =>
27 context.ask(blockchainActor, ref => GetLastHashEvent(ref)) {
28 case Success(prevHash) =>
29 obtainLastIndexEvent(transactions, prevHash)
30 case Failure(ex) =>
31 MiningFailed(ex.getMessage)
32 }
33 Behaviors.same
34
35 case obtainLastIndexEvent(transactions, prevHash) =>
36 context.ask(blockchainActor, ref => getIndexEvent(ref)) {
37 case Success(index) =>
38 mineBlock(transactions, prevHash, index)
39 case Failure(ex) =>
40 MiningFailed(ex.getMessage)
41 }
42 Behaviors.same
43
44 case mineBlock(transactions, prevHash, index) =>
45 context.log.info(
46 s"Mining block with ${transactions.size} transactions, prevHash=${prevHash}, index=${index}"
47 )
48 val (hash, proof, timestamp) = BlockDurable.mineBlock(index, prevHash, transactions)
49 val newBlock = Block(index, hash, prevHash, proof, timestamp, transactions)
50 blockchainActor ! AddBlockEvent(newBlock)
51 brokerActor ! ClearTransactionEvent
52 Behaviors.same
53
54 case MiningFailed(error) =>
55 context.log.error(s"Mining failed: $error")
56 Behaviors.same
57 }
58 }
59 }
60}Configuring a Durable Blockchain Actor
In this section, we implement the blockchain actor, which is responsible for maintaining the current state of our blockchain. A blockchain is essentially a list of blocks, and each block contains a group of transactions.
The actor responds to several key messages:
- GetChainEvent: retrieves the current list of blocks.
- AddBlockEvent: adds a new block mined locally to the blockchain.
- AddNeighbourBlockEvent: adds a block mined by a peer node (after validation).
- GetLastHashEvent: returns the hash of the last block.
- getIndexEvent: returns the index of the next block to be mined.
1sealed trait BlockChainEvent
2
3case class GetChainEvent(replyTo: ActorRef[List[Block]]) extends BlockChainEvent
4case class AddBlockEvent(block: Block) extends BlockChainEvent
5case class AddNeighbourBlockEvent(block: Block) extends BlockChainEvent
6case class GetLastHashEvent(replyTo: ActorRef[String]) extends BlockChainEvent
7case class getIndexEvent(replyTo: ActorRef[Int]) extends BlockChainEvent
8
9case class BlockChainState(blocks: List[Block]) extends SerializableWhen the blockchain actor is initialized, it must contain a genesis block—the first block in the chain. This is created using Block.createGenesisBlock(). We also pass in a reference to the network actor, which is used to broadcast newly mined blocks to peer nodes.
1def apply(nodeId: String, network: ActorRef[NetworkEvent]): Behavior[BlockChainEvent] = {
2 val genesisBlock = BlockDurable.createGenesisBlock()
3 blockchainDurableBehavior(List(genesisBlock), nodeId, network)
4}To ensure that our blockchain state persists across application restarts, we use Akka's DurableStateBehavior. This allows us to persist the blockchain data without relying on snapshots or event sourcing.
The DurableStateBehavior requires a few key components to function correctly. It needs a unique persistenceId to distinguish each actor instance, an initial state to start from—which typically includes the genesis block—and a command handler that determines how incoming messages should be processed.
The command handler defines how the actor responds to various events. When it receives a GetChainEvent, it replies with the current list of blocks. For an AddBlockEvent, it validates the incoming block, broadcasts it to the network, and appends it to the blockchain if it's valid. A AddNeighbourBlockEvent is similarly validated but appended silently without broadcasting. The getIndexEvent returns the index of the next block to be mined, while GetLastHashEvent provides the hash of the most recently added block.
1private def blockchainDurableBehavior(
2 blocks: List[Block],
3 nodeId: String,
4 network: ActorRef[NetworkEvent]
5): Behavior[BlockChainEvent] = {
6 Behaviors.setup { context =>
7 DurableStateBehavior[BlockChainEvent, BlockChainState](
8 persistenceId = PersistenceId.ofUniqueId(nodeId),
9 emptyState = BlockChainState(blocks),
10 commandHandler = (state, command) => command match {
11 case GetChainEvent(replyTo) =>
12 Effect.reply(replyTo)(state.blocks)
13
14 case AddBlockEvent(block) =>
15 val valid = BlockDurable.isValidProof(block)
16 if (valid) {
17 context.log.info("Block is valid, added")
18 network ! broadcastBlockEvent(block, nodeId)
19 Effect.persist(state.copy(blocks = state.blocks :+ block))
20 } else {
21 context.log.info("Block is invalid, rejected")
22 Effect.none
23 }
24
25 case AddNeighbourBlockEvent(block) =>
26 val valid = BlockDurable.isValidProof(block)
27 if (valid) {
28 context.log.info("Neighbour block is valid, added")
29 Effect.persist(state.copy(blocks = state.blocks :+ block))
30 } else {
31 context.log.info("Block is invalid, rejected")
32 Effect.none
33 }
34
35 case getIndexEvent(replyTo) =>
36 val lastBlock = state.blocks.last
37 Effect.reply(replyTo)(lastBlock.index + 1)
38
39 case GetLastHashEvent(replyTo) =>
40 val lastBlock = state.blocks.last
41 Effect.reply(replyTo)(lastBlock.hash)
42 }
43 )
44 }
45}Building a Storage Backend for Our Blockchain State
While Akka Persistence supports persisting state in relational databases (via R2DBC), Cassandra, or AWS DynamoDB, in this project we choose to persist the state locally using files. To achieve this, we implement a custom durable state store plugin by providing our own DurableStateUpdateStore. In our application.conf, we configure Akka to use the state.file plugin and specify a directory to store actor state files.
1akka {
2 persistence {
3 state.plugin = "akka.persistence.state.file"
4 state.file {
5 dir = "target/durable-state"
6 class = "durable.provider.FileDurableStateStoreProvider"
7 }
8 }
9}Next, we define our custom implementation of FileDurableStateStoreProvider, which extends DurableStateStoreProvider and provides both Scala and Java durable state stores. In this article, we focus on the Scala implementation.
1package durable.provider
2
3class FileDurableStateStoreProvider(system: ExtendedActorSystem, config: Config, cfgPath: String)
4 extends DurableStateStoreProvider {
5
6 override def scaladslDurableStateStore(): DurableStateStore[Any] =
7 new FileDurableStateStore(system, config, cfgPath)
8
9 override def javadslDurableStateStore(): javadsl.DurableStateStore[AnyRef] =
10 new JavaFileDurableStateStoreAdapter(system, config, cfgPath)
11}Our Scala implementation of FileDurableStateStore extends DurableStateUpdateStore. It must implement upsertObject, deleteObject, and getObject. The class reads the configured directory path, creates a state file per actor ID, and serializes the state into that file. When creating or updating the file, we ensure directories exist and use a combination of FileOutputStream and ObjectOutputStream to persist state. FileOutputStream is used to write raw bytes to a file, while ObjectOutputStream wraps around it to write full Scala (or Java) objects via serialization. Serialization is the process of converting an object into a byte stream for storage or transmission, and the object must implement Serializable. In our implementation, we write the revision, followed by a tag, and then the serialized object.
The deleteObject method deletes the file if it exists (and optionally matches a revision if provided). The getObject method reads back the file if present, retrieves the stored revision and object, and returns Some(value) wrapped in GetObjectResult with the appropriate revision. If the file does not exist, it returns an empty result and revision zero.
1class FileDurableStateStore[T](system: ExtendedActorSystem, config: Config, cfgPath: String)
2 extends DurableStateUpdateStore[T] {
3
4 private implicit val ec: ExecutionContext = system.dispatcher
5 private val baseDir: String = config.getString("dir") match {
6 case path if path.nonEmpty => path
7 case _ => "target/durable-state"
8 }
9
10 private def stateFile(persistenceId: String): File =
11 new File(s"$baseDir/$persistenceId.state")
12
13 override def upsertObject(persistenceId: String, revision: Long, value: T, tag: String): Future[Done] =
14 Future {
15 val file = stateFile(persistenceId)
16 file.getParentFile.mkdirs()
17 val oos = new ObjectOutputStream(new FileOutputStream(file))
18 try {
19 oos.writeLong(revision)
20 oos.writeUTF(tag)
21 oos.writeObject(value)
22 } finally {
23 oos.close()
24 }
25 Done
26 }
27
28 override def deleteObject(persistenceId: String): Future[Done] =
29 Future {
30 val file = stateFile(persistenceId)
31 if (file.exists()) file.delete()
32 Done
33 }
34
35 override def deleteObject(persistenceId: String, revision: Long): Future[Done] =
36 Future {
37 val file = stateFile(persistenceId)
38 if (file.exists()) {
39 val ois = new ObjectInputStream(new FileInputStream(file))
40 val storedRevision = try {
41 ois.readLong()
42 } finally {
43 ois.close()
44 }
45 if (storedRevision == revision) file.delete()
46 }
47 Done
48 }
49
50 override def getObject(persistenceId: String): Future[GetObjectResult[T]] =
51 Future {
52 val file = stateFile(persistenceId)
53 if (!file.exists()) {
54 GetObjectResult(None, 0L)
55 } else {
56 val ois = new ObjectInputStream(new FileInputStream(file))
57 try {
58 val revision = ois.readLong()
59 val tag = ois.readUTF()
60 val value = ois.readObject().asInstanceOf[T]
61 GetObjectResult(Some(value), revision)
62 } finally {
63 ois.close()
64 }
65 }
66 }
67}Implementing our node actor
In this section, we implement the Node actor, which is responsible for maintaining the blockchain actor, the broker actor, and the miner actor. It acts as the entry point for messages coming from the network and dispatches them to the appropriate internal components.
The actor responds to several key messages:
- AddNewTransactionEvent: adds a new transaction to the broker.
- GetTransactionsEvent: retrieves the list of unconfirmed transactions.
- WrappedTransactionsResponseEvent: sends the list of transactions to the actor reference provided.
- ReceiveNewBlockEvent: receives a block from neighbouring nodes.
- AppendBlockEvent: appends a block to the current blockchain.
- MineEvent: triggers mining of the current transactions.
- GetChainRequestEvent: fetches the current blockchain.
- GetChainResponseEvent: forwards the list of blocks to the requesting actor.
1sealed trait NodeEvent
2case class AddNewTransactionEvent(transaction: Transaction) extends NodeEvent
3case class GetTransactionsEvent(replyTo: ActorRef[List[Transaction]]) extends NodeEvent
4case class WrappedTransactionsResponseEvent(transactions: List[Transaction], replyTo: ActorRef[List[Transaction]]) extends NodeEvent
5case class ReceiveNewBlockEvent(block: Block) extends NodeEvent
6case class AppendBlockEvent(block: Block) extends NodeEvent
7case object MineEvent extends NodeEvent
8case class GetChainRequestEvent(replyTo: ActorRef[List[Block]]) extends NodeEvent
9case class GetChainResponseEvent(chain: List[Block], replyTo: ActorRef[List[Block]]) extends NodeEvent
10case class BlockchainReady() extends NodeEvent
11case class NodeError(ex: String) extends NodeEventIn the apply method, we pass the nodeId (which uniquely identifies each node) and a reference to the network actor so it can broadcast mined blocks.
1def apply(nodeId: String, network: ActorRef[NetworkEvent]): Behavior[NodeEvent] =
2 nodeBehavior(nodeId, network)In nodeBehavior, we spawn internal actors such as the blockchain, broker, and miner. The node coordinates communication between these components by forwarding or requesting events as needed. For instance, new transactions are forwarded to the broker, while mining instructions are sent to the miner. When a new block is received from a peer, it is validated before being appended to the local chain. Queries for transactions or the current blockchain state use the ask pattern, and the responses are wrapped in corresponding events like WrappedTransactionsResponseEvent or GetChainResponseEvent.
1object NodeDurable {
2 sealed trait NodeEvent
3 case class AddNewTransactionEvent(transaction: Transaction) extends NodeEvent
4 case class GetTransactionsEvent(replyTo: ActorRef[List[Transaction]]) extends NodeEvent
5 case class WrappedTransactionsResponseEvent(transactions: List[Transaction], replyTo: ActorRef[List[Transaction]]) extends NodeEvent
6 case class ReceiveNewBlockEvent(block: Block) extends NodeEvent
7 case class AppendBlockEvent(block: Block) extends NodeEvent
8 case object MineEvent extends NodeEvent
9 case class GetChainRequestEvent(replyTo: ActorRef[List[Block]]) extends NodeEvent
10 case class GetChainResponseEvent(chain: List[Block], replyTo: ActorRef[List[Block]]) extends NodeEvent
11 case class BlockchainReady() extends NodeEvent
12 case class NodeError(ex: String) extends NodeEvent
13
14 def apply(nodeId: String, network: ActorRef[NetworkEvent]): Behavior[NodeEvent] =
15 nodeBehavior(nodeId, network)
16
17 private def nodeBehavior(nodeId: String, network: ActorRef[NetworkEvent]): Behavior[NodeEvent] = {
18 Behaviors.setup { context =>
19 implicit val timeout: Timeout = 3.seconds
20
21 val blockchainActor = context.spawn(BlockChainDurable(nodeId, network), "BlockChainDurable")
22 val brokerActor = context.spawn(BrokerDurable(), "brokerDurable")
23 val minerActor = context.spawn(MinerDurable(brokerActor, blockchainActor), "minerDurable")
24
25 Behaviors.receive { (context, message) =>
26 message match {
27 case AddNewTransactionEvent(transaction) =>
28 brokerActor ! AddTransactionEvent(transaction)
29 Behaviors.same
30
31 case GetTransactionsEvent(replyTo) =>
32 context.ask(brokerActor, ref => GetTransactionEvent(ref)) {
33 case Success(transactions) => WrappedTransactionsResponseEvent(transactions, replyTo)
34 case Failure(ex) => NodeError("get transactions: " + ex.getMessage)
35 }
36 Behaviors.same
37
38 case WrappedTransactionsResponseEvent(transactions, replyTo) =>
39 replyTo ! transactions
40 Behaviors.same
41
42 case ReceiveNewBlockEvent(block) =>
43 context.log.info(s"$nodeId received a new block from neighbours")
44 context.ask(minerActor, ref => ValidateBlock(block, ref)) {
45 case Success(true) => AppendBlockEvent(block)
46 case Success(false) => NodeError("Block received is invalid")
47 case Failure(ex) => NodeError(ex.getMessage)
48 }
49 Behaviors.same
50
51 case AppendBlockEvent(block) =>
52 blockchainActor ! AddNeighbourBlockEvent(block)
53 Behaviors.same
54
55 case MineEvent =>
56 minerActor ! MineCurrentBlockMinerEvent
57 Behaviors.same
58
59 case GetChainRequestEvent(replyTo) =>
60 context.ask(blockchainActor, ref => GetChainEvent(ref)) {
61 case Success(blocks) => GetChainResponseEvent(blocks, replyTo)
62 case Failure(ex) => NodeError(ex.getMessage)
63 }
64 Behaviors.same
65
66 case GetChainResponseEvent(blocks, replyTo) =>
67 replyTo ! blocks
68 Behaviors.same
69
70 case NodeError(error) =>
71 context.log.error(error)
72 Behaviors.same
73 }
74 }
75 }
76 }
77}Defining our network
In this section, we implement the network actor, which is responsible for managing the communication between nodes in the system. It simulates a gossip protocol by forwarding newly mined blocks to all other nodes in the network.
The network actor handles two primary message types:
- broadcastBlockEvent: takes in a Block and the sender's nodeId, and broadcasts the block to all other nodes in the network—excluding the sender.
- SendToNode: used for targeted messaging, where the network forwards a NodeEvent to a specific node.
1sealed trait NetworkEvent
2case class broadcastBlockEvent(block: Block, nodeId: String) extends NetworkEvent
3case class SendToNode(nodeId: String, event: NodeEvent) extends NetworkEvent
4
5def apply(): Behavior[NetworkEvent] =
6 networkBehavior()To construct the network, we first spawn the individual node actors—in this example, two: node1 and node2. We then maintain both a Map and a List to manage references to these nodes. The Map[String, ActorRef] allows us to efficiently look up the actor reference of a node for direct messaging (via SendToNode). The List[(String, ActorRef)] is used during block broadcasting so that we can easily filter out the sender and forward the block to the rest.
During broadcasting, the network filters out the node that originally sent the block, preventing it from receiving the same block again. The remaining peers each receive the ReceiveNewBlockEvent, enabling propagation of the new block across the network. For direct node communication, the SendToNode message retrieves the intended node actor from the map and sends it the event.
1private def networkBehavior(): Behavior[NetworkEvent] = {
2 Behaviors.setup { context =>
3 implicit val timeout: Timeout = 3.seconds
4 val node1 = context.spawn(NodeDurable("node1", context.self), "node1")
5 val node2 = context.spawn(NodeDurable("node2", context.self), "node2")
6 val nodeMap = Map("node1" -> node1, "node2" -> node2)
7 val peers = List("node1" -> node1, "node2" -> node2)
8 Behaviors.receive { (context, message) =>
9 message match {
10 case broadcastBlockEvent(block, senderId) =>
11 context.log.info(s"Received broadcasting event from $senderId")
12 val broadcast = peers.filterNot {
13 case (id, _) => id == senderId
14 }
15 broadcast.foreach {
16 case (_, actor) => actor ! ReceiveNewBlockEvent(block)
17 }
18 Behaviors.same
19
20 case SendToNode(nodeId, event) =>
21 nodeMap.get(nodeId).foreach(_ ! event)
22 Behaviors.same
23 }
24 }
25 }
26}Exposing Our Network via Akka HTTP
In this section, we expose our blockchain network using Akka HTTP. We provide three main endpoints:
1. Add a transaction to a specific node
2. Trigger mining on a specific node
3. Retrieve the blockchain from a specific node
We begin by initializing our actor system using the Network actor, and set up the necessary Akka infrastructure such as the execution context, scheduler, and timeout. We also define a case class AddTransaction and its corresponding jsonFormat for deserializing incoming JSON requests.
1implicit val system : ActorSystem[Network.NetworkEvent] = ActorSystem(Network(), "network")
2implicit val executionContext = system.executionContext
3implicit val timeout:Timeout = 3.seconds
4implicit val scheduler = system.scheduler
5
6final case class AddTransaction(nodeId: String,sender: String, receiver: String, amount: Int)
7implicit val addTransactionFormat: RootJsonFormat[AddTransaction] = jsonFormat4(AddTransaction.apply)Additionally, we define RootJsonFormat serializers for our Transaction and Block types within BlockDurable.scala. These will be used to correctly marshal and unmarshal JSON payloads when interacting with the HTTP endpoints.
1// Under BlockDurable.scala
2object JsonFormats {
3 implicit val transactionFormat: RootJsonFormat[Transaction] = jsonFormat3(Transaction)
4 implicit val blockFormat: RootJsonFormat[Block] = jsonFormat6(Block)
5}We then define the routes for our Akka HTTP server:
/transaction: Accepts a POST request with transaction data, deserializes it into an AddTransaction object, and sends an AddNewTransactionEvent to the specified node via the SendToNode message.
/mine: Accepts a GET request with a nodeId query parameter, and triggers a MineEvent on that node.
/chain/{nodeId}: Retrieves the blockchain for a given node by sending a GetChainRequestEvent using the ask pattern. The resulting Future[List[Block]] is then serialized to JSON and returned.
1val route: Route =
2 concat(
3 post {
4 path("transaction") {
5 entity(as[AddTransaction]) { transaction =>
6 system ! SendToNode(transaction.nodeId,
7 AddNewTransactionEvent(Transaction(transaction.sender, transaction.receiver, transaction.amount)))
8 complete("transaction created")
9 }
10 }
11 },
12 get {
13 path("mine") {
14 parameter("nodeId") { nodeId =>
15 system ! SendToNode(nodeId, MineEvent)
16 complete("Block is mined")
17 }
18 }
19 },
20 get {
21 pathPrefix("chain" / Segment) { nodeId =>
22 val futureChain: Future[List[Block]] =
23 system.ask(replyTo => SendToNode(nodeId, GetChainRequestEvent(replyTo)))
24
25 onComplete(futureChain) {
26 case Success(chain) =>
27 complete(chain)
28 case Failure(ex) =>
29 complete(StatusCodes.InternalServerError, s"Failed to get blockchain: ${ex.getMessage}")
30 }
31 }
32 }
33 )
34
35val bindingFuture = Http().newServerAt("localhost", 8080).bind(route)
36
37println("Server now online. Please navigate to http://localhost:8080")
38StdIn.readLine()
39bindingFuture
40 .flatMap(_.unbind())
41 .onComplete(_ => system.terminate())