add GetNodeData tests for stream client, increase nodes and receipts cap

p2p/stream/protocols/sync/chain_test.go

@@ -75,6 +75,21 @@ func checkGetReceiptsResult(b []byte, hs []common.Hash) error {
 	return nil
 }
 
+func checkGetNodeDataResult(b []byte, hs []common.Hash) error {
+	var msg = &syncpb.Message{}
+	if err := protobuf.Unmarshal(b, msg); err != nil {
+		return err
+	}
+	bhResp, err := msg.GetNodeDataResponse()
+	if err != nil {
+		return err
+	}
+	if len(hs) != len(bhResp.DataBytes) {
+		return errors.New("unexpected size")
+	}
+	return nil
+}
+
 func numberToHash(bn uint64) common.Hash {
 	var h common.Hash
 	binary.LittleEndian.PutUint64(h[:], bn)

p2p/stream/protocols/sync/const.go

@@ -19,11 +19,11 @@ const (
 
 	// GetNodeDataCap is the cap of request of single GetNodeData request
 	// This number has an effect on maxMsgBytes as 20MB defined in github.com/harmony-one/harmony/p2p/stream/types.
-	GetNodeDataCap = 10
+	GetNodeDataCap = 256
 
 	// GetReceiptsCap is the cap of request of single GetReceipts request
 	// This number has an effect on maxMsgBytes as 20MB defined in github.com/harmony-one/harmony/p2p/stream/types.
-	GetReceiptsCap = 10
+	GetReceiptsCap = 128
 
 	// MaxStreamFailures is the maximum allowed failures before stream gets removed
 	MaxStreamFailures = 5

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

@@ -95,3 +95,19 @@ func (msg *Message) GetReceiptsResponse() (*GetReceiptsResponse, error) {
 	}
 	return grResp, nil
 }
+
+// GetNodeDataResponse parse the message to GetNodeDataResponse
+func (msg *Message) GetNodeDataResponse() (*GetNodeDataResponse, 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}
+	}
+	gnResp := resp.GetGetNodeDataResponse()
+	if gnResp == nil {
+		return nil, errors.New("not GetGetNodeDataResponse")
+	}
+	return gnResp, nil
+}

p2p/stream/protocols/sync/stream_test.go

@@ -50,6 +50,16 @@ var (
 	}
 	testGetReceiptsRequest    = syncpb.MakeGetReceiptsRequest(testGetReceipts)
 	testGetReceiptsRequestMsg = syncpb.MakeMessageFromRequest(testGetReceiptsRequest)
+
+	testGetNodeData = []common.Hash{
+		numberToHash(1),
+		numberToHash(2),
+		numberToHash(3),
+		numberToHash(4),
+		numberToHash(5),
+	}
+	testGetNodeDataRequest    = syncpb.MakeGetNodeDataRequest(testGetNodeData)
+	testGetNodeDataRequestMsg = syncpb.MakeMessageFromRequest(testGetNodeDataRequest)
 )
 
 func TestSyncStream_HandleGetBlocksByRequest(t *testing.T) {
@@ -152,7 +162,28 @@ func TestSyncStream_HandleGetReceipts(t *testing.T) {
 	time.Sleep(200 * time.Millisecond)
 	receivedBytes, _ := remoteSt.ReadBytes()
 
-	if err := checkGetReceiptsResult(receivedBytes, testGetBlockByHashes); err != nil {
+	if err := checkGetReceiptsResult(receivedBytes, testGetReceipts); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestSyncStream_HandleGetNodeData(t *testing.T) {
+	st, remoteSt := makeTestSyncStream()
+
+	go st.run()
+	defer close(st.closeC)
+
+	req := testGetNodeDataRequestMsg
+	b, _ := protobuf.Marshal(req)
+	err := remoteSt.WriteBytes(b)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	time.Sleep(200 * time.Millisecond)
+	receivedBytes, _ := remoteSt.ReadBytes()
+
+	if err := checkGetNodeDataResult(receivedBytes, testGetNodeData); err != nil {
 		t.Fatal(err)
 	}
 }