talk is cheap, show me the code!
package main
import (
"context"
"crypto/sha256"
"encoding/json"
"flag"
"fmt"
"io/ioutil"
"net"
"net/http"
"strings"
"time"
"github.com/gin-gonic/gin"
log "github.com/laohanlinux/utils/gokitlog"
"github.com/laohanlinux/utils/netrpc"
pool "github.com/laohanlinux/utils/pool/rpc"
)
var (
blockChain = []Block{GetGenesiseBlock()}
initialPeers = flag.String("peers", "localhost:6660,localhost:6661,localhost:6662", "")
httpPort = flag.String("http_port", "localhost:7000", "")
p2pPort = flag.String("p2p_port", "localhost:6660", "")
peersConnections = make(map[string]*pool.NetRPCRing)
s Service
)
func main() {
flag.Parse()
l, err := initP2PServer()
if err != nil {
log.Crit("err", err)
}
log.Debug("p2pServer listener address", *p2pPort)
defer l.Close()
peers := strings.Split(*initialPeers, ",")
conncetToPeers(peers)
initHttpServer()
time.Sleep(time.Hour * 10000)
}
type Block struct {
Index int64
PreviousHash string
TimeStamp int64
Data []byte
Hash string
}
func NewBlock(index int64, previousHash string, timeStamp int64, Data []byte, Hash string) Block {
return Block{
Index: index,
PreviousHash: previousHash,
TimeStamp: timeStamp,
Data: Data,
Hash: Hash,
}
}
func CalculationHashForBlock(block Block) string {
return CalculationHash(block.Index, block.PreviousHash, block.TimeStamp, block.Data)
}
func CalculationHash(index int64, previousHash string, timeStamp int64, data []byte) string {
return fmt.Sprintf("%x", sha256.Sum256([]byte(fmt.Sprintf("%v%v%v%s", index, previousHash, timeStamp, data))))
}
func GenerateNextBlock(blockData []byte) Block {
var (
previousBlock = getLatestBlock()
nextIndex = previousBlock.Index + 1
nextTimeStamp = time.Now().Unix()
nextHash = CalculationHash(nextIndex, previousBlock.Hash, nextTimeStamp, blockData)
)
return NewBlock(nextIndex, previousBlock.Hash, nextTimeStamp, blockData, nextHash)
}
func GetGenesiseBlock() Block {
return NewBlock(0, "0", 1465154705, []byte("my genesis block!!"), "816534932c2b7154836da6afc367695e6337db8a921823784c14378abed4f7d7")
}
func AddBlock(newBlock Block) {
if isValidNewBlock(newBlock, getLatestBlock()) {
blockChain = append(blockChain, newBlock)
}
}
func replaceChain(newBlocks []Block) {
if isValidBlock(newBlocks) && len(newBlocks) > len(blockChain) {
log.Warnf("received blockchain is valid, replacing current blockchain with received blockchain")
blockChain = newBlocks
// broadcast
for _, peerClient := range peersConnections {
go peerClient.Call(serviceAddBlockChain, &AddBlockChainArgs{NewBlockChain: blockChain}, &AddBlockChainReply{})
}
} else {
log.Debugf("received blockchain invalid")
}
}
func isValidNewBlock(newBlock, previousBlock Block) bool {
if previousBlock.Index+1 != newBlock.Index {
log.Debug("msg", "invalid index")
return false
}
if previousBlock.Hash != newBlock.PreviousHash {
log.Debug("msg", "invalid previousHash")
return false
}
// check wether newBlock.Hash is right calculation by his content
if CalculationHashForBlock(newBlock) != newBlock.Hash {
log.Debugf("invalid hash: %v, %v", CalculationHashForBlock(newBlock), newBlock.Hash)
return false
}
return true
}
func isValidBlock(blockChainToValidate []Block) bool {
// check genesis block
if blockChainToValidate[0].Hash != GetGenesiseBlock().Hash {
return false
}
for i := 1; i < len(blockChainToValidate); i++ {
if !isValidNewBlock(blockChainToValidate[i], blockChainToValidate[i-1]) {
return false
}
}
return true
}
func getLatestBlock() Block { return blockChain[len(blockChain)-1] }
func initHttpServer() {
router := gin.Default()
router.GET("/blocks", func(c *gin.Context) { json.NewEncoder(c.Writer).Encode(blockChain) })
router.POST("/mineBlock", func(c *gin.Context) {
var (
buf []byte
err error
)
if buf, err = ioutil.ReadAll(c.Request.Body); err != nil {
log.Error("err", err)
c.Writer.WriteHeader(http.StatusInternalServerError)
return
}
AddBlock(GenerateNextBlock(buf))
// broadcast to notify peers our has add new block
for _, peerClient := range peersConnections {
go func() {
var rpcArgs, rpcReply = AddBlockChainArgs{NewBlockChain: blockChain}, AddBlockChainReply{}
if err := peerClient.Call(serviceAddBlockChain, &rpcArgs, &rpcReply); err != nil {
log.Error("err", err)
}
}()
}
c.Writer.WriteHeader(http.StatusOK)
})
router.GET("/peers", func(c *gin.Context) {
var peers []string
for peer, _ := range peersConnections {
peers = append(peers, peer)
}
if err := json.NewEncoder(c.Writer).Encode(peers); err != nil {
log.Error("err", err)
}
})
router.POST("/addPeer", func(c *gin.Context) {
var peers []string
if err := json.NewDecoder(c.Request.Body).Decode(&peers); err != nil {
log.Error("err", err)
c.Writer.WriteHeader(http.StatusBadRequest)
return
}
conncetToPeers(peers)
})
go http.ListenAndServe(*httpPort, router)
}
func initP2PServer() (net.Listener, error) {
l, err := net.Listen("tcp", *p2pPort)
if err != nil {
return nil, err
}
server := netrpc.NewServer()
server.Register(&s)
server.Register(&netrpc.HealthCheck{})
go server.Accept(l)
return l, nil
}
func conncetToPeers(peers []string) {
var (
rpcOpt pool.NetRPCRingOpt
)
for _, peer := range peers {
if _, ok := peersConnections[peer]; ok {
continue
}
rpcOpt.Addr, rpcOpt.NetWork, rpcOpt.PoolSize = peer, "tcp", 1
if pools, err := pool.NewNetRPCRing([]pool.NetRPCRingOpt{rpcOpt}); err != nil {
log.Error("err", err)
} else {
peersConnections[peer] = pools
}
}
}
// AddBlockChain add new blockchain into cluster
// if the new block is new than local lasted block, agree it
func handleBlockChain(args AddBlockChainArgs) {
log.Debug("who", *p2pPort, "do", "handle blockchain")
var (
receivedBlocks = args.NewBlockChain
latestBlockReceived = receivedBlocks[len(receivedBlocks)-1]
latestBlockHeld = getLatestBlock()
)
if latestBlockReceived.Index > latestBlockHeld.Index {
log.Warnf("blockchain possibly behind. We got: %v Peer got: %v", latestBlockHeld.Index, latestBlockReceived.Index)
if latestBlockHeld.Hash == latestBlockReceived.PreviousHash {
log.Warnf("We can append the received block to our block")
blockChain = append(blockChain, latestBlockReceived)
// broadcast to other
for _, peerClient := range peersConnections {
go func() {
var rpcArgs, rpcReply = AddBlockChainArgs{NewBlockChain: args.NewBlockChain}, AddBlockChainReply{}
if err := peerClient.Call(serviceAddBlockChain, &rpcArgs, &rpcReply); err != nil {
log.Error("err", err)
}
}()
}
} else if len(receivedBlocks) == 1 {
log.Warnf("we have must query the chain from our peer")
// broadcast
for _, peersClient := range peersConnections {
go func() {
var (
rpcArgs QueryAllBlockChainsArgs
rpcreply QueryAllBlockChainsReply
err error
)
if err = peersClient.Call(serviceQueryAllBlockChains, &rpcArgs, &rpcreply); err != nil {
log.Error("err", err)
} else {
handleBlockChain(AddBlockChainArgs{NewBlockChain: rpcreply.NewBlockChains})
}
}()
}
} else {
log.Debugf("Received blockchain is longer than current blockchain")
replaceChain(receivedBlocks)
}
} else {
log.Debug("received blockchain is not longer than received blockchain. Do nothing")
}
}
const (
serviceName = "Service"
serviceAddBlockChain = serviceName + ".AddBlockChain"
serviceQueryAllBlockChains = serviceName + ".QueryAllBlockChains"
)
type AddBlockChainArgs struct{ NewBlockChain []Block }
type AddBlockChainReply struct{}
///////
type QueryAllBlockChainsArgs struct{}
type QueryAllBlockChainsReply struct{ NewBlockChains []Block }
//////
type Service struct{}
func (s *Service) QueryAllMsg(_ context.Context, _ *QueryAllBlockChainsArgs, reply *QueryAllBlockChainsReply) error {
reply.NewBlockChains = blockChain
return nil
}
func (s *Service) AddBlockChain(_ context.Context, args *AddBlockChainArgs, reply *AddBlockChainReply) error {
handleBlockChain(*args)
return nil
}
启动脚本：
go build -o blockchain main.go
sudo pkill -9 blockchain
./blockchain -http_port="localhost:7000" -p2p_port="localhost:6660" -peers="localhost:6661" &
./blockchain -http_port="localhost:7001" -p2p_port="localhost:6661" -peers="localhost:6660" &
sleep 1
curl -v -XPOST "http://localhost:7001/addPeer" -d '[localhost:6660]' | jq
sleep 2
curl -v -XPOST "http://localhost:7000/addPeer" -d '[localhost:6661]' | jq
sleep 2
curl -v -XPOST "http://localhost:7001/mineBlock" -d 'hello word' | jq
sleep 1
curl -v -XPOST "http://localhost:7001/mineBlock" -d 'hello word' | jq
sleep 1
curl -v -XPOST "http://localhost:7001/mineBlock" -d 'hello word' | jq
sleep 1
curl -v -XPOST "http://localhost:7001/mineBlock" -d 'hello word' | jq
sleep 1
curl -v -XPOST "http://localhost:7001/mineBlock" -d 'hello word' | jq
sleep 1
curl -v -XPOST "http://localhost:7001/mineBlock" -d 'hello word' | jq
sleep 1
curl -v -XPOST "http://localhost:7001/mineBlock" -d 'hello word' | jq
sleep 1
curl -v "http://localhost:7001/blocks" | jq
sleep 1
curl -v "http://localhost:7000/blocks" | jq
curl -v "http://localhost:7000/peers" | jq
curl -v "http://localhost:7001/peers" | jq
比较简单，拿来看看就好了，别太在意….