[stream] added sync stream protobuf message module (#3561)

Co-authored-by: Rongjian Lan <rongjian.lan@gmail.com>

p2p/stream/protocols/sync/message/compose.go

@@ -0,0 +1,206 @@
+package message
+
+import (
+	"github.com/ethereum/go-ethereum/common"
+)
+
+// MakeGetBlockNumberRequest makes the GetBlockNumber Request
+func MakeGetBlockNumberRequest() *Request {
+	return &Request{
+		Request: &Request_GetBlockNumberRequest{
+			GetBlockNumberRequest: &GetBlockNumberRequest{},
+		},
+	}
+}
+
+// MakeGetBlockHashesRequest makes GetBlockHashes Request
+func MakeGetBlockHashesRequest(bns []uint64) *Request {
+	return &Request{
+		Request: &Request_GetBlockHashesRequest{
+			GetBlockHashesRequest: &GetBlockHashesRequest{
+				Nums: bns,
+			},
+		},
+	}
+}
+
+// MakeGetBlocksByNumRequest makes the GetBlockByNumber request
+func MakeGetBlocksByNumRequest(bns []uint64) *Request {
+	return &Request{
+		Request: &Request_GetBlocksByNumRequest{
+			GetBlocksByNumRequest: &GetBlocksByNumRequest{
+				Nums: bns,
+			},
+		},
+	}
+}
+
+//MakeGetBlockByHashesRequest makes the GetBlocksByHashes request
+func MakeGetBlocksByHashesRequest(hashes []common.Hash) *Request {
+	return &Request{
+		Request: &Request_GetBlocksByHashesRequest{
+			GetBlocksByHashesRequest: &GetBlocksByHashesRequest{
+				BlockHashes: hashesToBytes(hashes),
+			},
+		},
+	}
+}
+
+// MakeGetEpochStateRequest make GetEpochBlock request
+func MakeGetEpochStateRequest(epoch uint64) *Request {
+	return &Request{
+		Request: &Request_GetEpochStateRequest{
+			GetEpochStateRequest: &GetEpochStateRequest{
+				Epoch: epoch,
+			},
+		},
+	}
+}
+
+// MakeErrorResponse makes the error response
+func MakeErrorResponseMessage(rid uint64, err error) *Message {
+	resp := MakeErrorResponse(rid, err)
+	return makeMessageFromResponse(resp)
+}
+
+// MakeErrorResponse makes the error response as a response
+func MakeErrorResponse(rid uint64, err error) *Response {
+	return &Response{
+		ReqId: rid,
+		Response: &Response_ErrorResponse{
+			&ErrorResponse{
+				Error: err.Error(),
+			},
+		},
+	}
+}
+
+// MakeGetBlockNumberResponseMessage makes the GetBlockNumber response message
+func MakeGetBlockNumberResponseMessage(rid uint64, bn uint64) *Message {
+	resp := MakeGetBlockNumberResponse(rid, bn)
+	return makeMessageFromResponse(resp)
+}
+
+// MakeGetBlockNumberResponse makes the GetBlockNumber response
+func MakeGetBlockNumberResponse(rid uint64, bn uint64) *Response {
+	return &Response{
+		ReqId: rid,
+		Response: &Response_GetBlockNumberResponse{
+			GetBlockNumberResponse: &GetBlockNumberResponse{
+				Number: bn,
+			},
+		},
+	}
+}
+
+// MakeGetBlockHashesResponseMessage makes the GetBlockHashes message
+func MakeGetBlockHashesResponseMessage(rid uint64, hs []common.Hash) *Message {
+	resp := MakeGetBlockHashesResponse(rid, hs)
+	return makeMessageFromResponse(resp)
+}
+
+// MakeGetBlockHashesResponse makes the GetBlockHashes response
+func MakeGetBlockHashesResponse(rid uint64, hs []common.Hash) *Response {
+	return &Response{
+		ReqId: rid,
+		Response: &Response_GetBlockHashesResponse{
+			GetBlockHashesResponse: &GetBlockHashesResponse{
+				Hashes: hashesToBytes(hs),
+			},
+		},
+	}
+}
+
+// MakeGetBlocksByNumResponseMessage makes the GetBlocksByNumResponse of Message type
+func MakeGetBlocksByNumResponseMessage(rid uint64, blocksBytes [][]byte) *Message {
+	resp := MakeGetBlocksByNumResponse(rid, blocksBytes)
+	return makeMessageFromResponse(resp)
+}
+
+// MakeGetBlocksByNumResponseMessage make the GetBlocksByNumResponse of Response type
+func MakeGetBlocksByNumResponse(rid uint64, blocksBytes [][]byte) *Response {
+	return &Response{
+		ReqId: rid,
+		Response: &Response_GetBlocksByNumResponse{
+			GetBlocksByNumResponse: &GetBlocksByNumResponse{
+				BlocksBytes: blocksBytes,
+			},
+		},
+	}
+}
+
+// MakeGetBlocksByHashesResponseMessage makes the GetBlocksByHashesResponse of Message type
+func MakeGetBlocksByHashesResponseMessage(rid uint64, blocksBytes [][]byte) *Message {
+	resp := MakeGetBlocksByHashesResponse(rid, blocksBytes)
+	return makeMessageFromResponse(resp)
+}
+
+// MakeGetBlocksByHashesResponse make the GetBlocksByHashesResponse of Response type
+func MakeGetBlocksByHashesResponse(rid uint64, blocksBytes [][]byte) *Response {
+	return &Response{
+		ReqId: rid,
+		Response: &Response_GetBlocksByHashesResponse{
+			GetBlocksByHashesResponse: &GetBlocksByHashesResponse{
+				BlocksBytes: blocksBytes,
+			},
+		},
+	}
+}
+
+// MakeGetEpochStateResponse makes GetEpochStateResponse as message
+func MakeGetEpochStateResponseMessage(rid uint64, headerBytes []byte, ssBytes []byte) *Message {
+	resp := MakeGetEpochStateResponse(rid, headerBytes, ssBytes)
+	return makeMessageFromResponse(resp)
+}
+
+// MakeEpochStateResponse makes GetEpochStateResponse as response
+func MakeGetEpochStateResponse(rid uint64, headerBytes []byte, ssBytes []byte) *Response {
+	return &Response{
+		ReqId: rid,
+		Response: &Response_GetEpochStateResponse{
+			GetEpochStateResponse: &GetEpochStateResponse{
+				HeaderBytes: headerBytes,
+				ShardState:  ssBytes,
+			},
+		},
+	}
+}
+
+// MakeMessageFromRequest makes a message from the request
+func MakeMessageFromRequest(req *Request) *Message {
+	return &Message{
+		ReqOrResp: &Message_Req{
+			Req: req,
+		},
+	}
+}
+
+func makeMessageFromResponse(resp *Response) *Message {
+	return &Message{
+		ReqOrResp: &Message_Resp{
+			Resp: resp,
+		},
+	}
+}
+
+func hashesToBytes(hashes []common.Hash) [][]byte {
+	res := make([][]byte, 0, len(hashes))
+
+	for _, h := range hashes {
+		b := make([]byte, common.HashLength)
+		copy(b, h[:])
+		res = append(res, b)
+	}
+	return res
+}
+
+func bytesToHashes(bs [][]byte) []common.Hash {
+	res := make([]common.Hash, len(bs))
+
+	for _, b := range bs {
+		var h common.Hash
+		copy(h[:], b)
+		res = append(res, h)
+	}
+	return res
+}

p2p/stream/protocols/sync/message/generate.go

@@ -0,0 +1,3 @@
+package message
+
+//go:generate protoc msg.proto --go_out=.

p2p/stream/protocols/sync/message/msg.proto

@@ -0,0 +1,80 @@
+syntax = "proto3";
+package harmony.stream.sync.message ;
+
+option go_package = "message";
+
+message Message {
+  oneof req_or_resp {
+    Request req = 1;
+    Response resp = 2;
+  }
+}
+
+message Request {
+  uint64 req_id = 1;
+  oneof request {
+    GetBlockNumberRequest get_block_number_request = 2;
+    GetBlockHashesRequest get_block_hashes_request = 3;
+    GetBlocksByNumRequest get_blocks_by_num_request = 4;
+    GetBlocksByHashesRequest get_blocks_by_hashes_request = 5;
+    GetEpochStateRequest get_epoch_state_request = 6;
+  }
+}
+
+message GetBlockNumberRequest {}
+
+message GetBlockHashesRequest {
+  repeated uint64 nums = 1 [packed=true];
+}
+
+message GetBlocksByNumRequest {
+  repeated uint64 nums = 1 [packed=true];
+}
+
+message GetBlocksByHashesRequest {
+  repeated bytes block_hashes = 1;
+}
+
+message GetEpochStateRequest {
+  uint64 epoch = 1;
+}
+
+message Response {
+  uint64 req_id = 1;
+  oneof response {
+    ErrorResponse error_response = 2;
+    GetBlockNumberResponse get_block_number_response = 3;
+    GetBlockHashesResponse get_block_hashes_response = 4;
+    GetBlocksByNumResponse get_blocks_by_num_response = 5;
+    GetBlocksByHashesResponse get_blocks_by_hashes_response = 6;
+    GetEpochStateResponse get_epoch_state_response = 7;
+  }
+}
+
+message ErrorResponse {
+  string error = 1;
+}
+
+message GetBlockNumberResponse {
+  uint64 number = 1;
+}
+
+message GetBlockHashesResponse {
+  repeated bytes hashes = 1;
+}
+
+message GetBlocksByNumResponse {
+  repeated bytes blocks_bytes = 1;
+}
+
+message GetBlocksByHashesResponse {
+  repeated bytes blocks_bytes = 1;
+}
+
+message GetEpochStateResponse {
+  bytes header_bytes = 1;
+  bytes shard_state = 2;
+}
+
+
+

p2p/stream/protocols/sync/message/parse.go

@@ -0,0 +1,97 @@
+package message
+
+import (
+	"fmt"
+
+	"github.com/pkg/errors"
+)
+
+// ResponseError is the error from an error response
+type ResponseError struct {
+	msg string
+}
+
+// Error is the error string of ResponseError
+func (err *ResponseError) Error() string {
+	return fmt.Sprintf("[RESPONSE] %v", err.msg)
+}
+
+// GetBlockNumberResponse parse the message to GetBlockNumberResponse
+func (msg *Message) GetBlockNumberResponse() (*GetBlockNumberResponse, error) {
+	resp := msg.GetResp()
+	if resp == nil {
+		return nil, errors.New("not response message")
+	}
+	if errResp := resp.GetErrorResponse(); errResp != nil {
+		return nil, &ResponseError{errResp.Error}
+	}
+	bnResp := resp.GetGetBlockNumberResponse()
+	if bnResp == nil {
+		return nil, errors.New("not GetBlockNumber response")
+	}
+	return bnResp, nil
+}
+
+// GetBlockHashesResponse parse the message to GetBlockHashesResponse
+func (msg *Message) GetBlockHashesResponse() (*GetBlockHashesResponse, error) {
+	resp := msg.GetResp()
+	if resp == nil {
+		return nil, errors.New("not response message")
+	}
+	if errResp := resp.GetErrorResponse(); errResp != nil {
+		return nil, &ResponseError{errResp.Error}
+	}
+	ghResp := resp.GetGetBlockHashesResponse()
+	if ghResp == nil {
+		return nil, errors.New("not GetBlockHashesResponse")
+	}
+	return ghResp, nil
+}
+
+// GetBlocksByNumberResponse parse the message to GetBlocksByNumberResponse
+func (msg *Message) GetBlocksByNumberResponse() (*GetBlocksByNumResponse, error) {
+	resp := msg.GetResp()
+	if resp == nil {
+		return nil, errors.New("not response message")
+	}
+	if errResp := resp.GetErrorResponse(); errResp != nil {
+		return nil, &ResponseError{errResp.Error}
+	}
+	gbResp := resp.GetGetBlocksByNumResponse()
+	if gbResp == nil {
+		return nil, errors.New("not GetBlocksByNumResponse")
+	}
+	return gbResp, nil
+}
+
+// GetBlocksByHashesResponse parse the message to GetBlocksByHashesResponse
+func (msg *Message) GetBlocksByHashesResponse() (*GetBlocksByHashesResponse, error) {
+	resp := msg.GetResp()
+	if resp == nil {
+		return nil, errors.New("not response message")
+	}
+	if errResp := resp.GetErrorResponse(); errResp != nil {
+		return nil, &ResponseError{errResp.Error}
+	}
+	gbResp := resp.GetGetBlocksByHashesResponse()
+	if gbResp == nil {
+		return nil, errors.New("not GetBlocksByHashesResponse")
+	}
+	return gbResp, nil
+}
+
+// GetEpochStateResponse parse the message to GetEpochStateResponse
+func (msg *Message) GetEpochStateResponse() (*GetEpochStateResponse, error) {
+	resp := msg.GetResp()
+	if resp == nil {
+		return nil, errors.New("not response message")
+	}
+	if errResp := resp.GetErrorResponse(); errResp != nil {
+		return nil, &ResponseError{errResp.Error}
+	}
+	gesResp := resp.GetGetEpochStateResponse()
+	if gesResp == nil {
+		return nil, errors.New("not GetEpochStateResponse")
+	}
+	return gesResp, nil
+}