feat(net): introduce 2.0.0 protocol for dag-cbor

Author: rvagg

benchmarks/testnet/virtual.go

@@ -137,7 +137,7 @@ func (n *network) SendMessage(
 			rateLimiters[to] = rateLimiter
 		}
 
-		pbMsg, err := gsmsg.NewMessageHandler().ToProto(mes)
+		pbMsg, err := gsmsg.NewMessageHandler().ToProtoV11(mes)
 		if err != nil {
 			return err
 		}

message/bench_test.go

@@ -73,7 +73,7 @@ func BenchmarkMessageEncodingRoundtrip(b *testing.B) {
 			for pb.Next() {
 				buf.Reset()
 
-				ipldGSM, err := gsm.ToIPLD()
+				ipldGSM, err := NewMessageHandler().ToIPLD(gsm)
 				require.NoError(b, err)
 				node := bindnode.Wrap(ipldGSM, ipldbind.Prototype.Message.Type())
 				err = dagcbor.Encode(node.Representation(), buf)
@@ -84,7 +84,7 @@ func BenchmarkMessageEncodingRoundtrip(b *testing.B) {
 				require.NoError(b, err)
 				node2 := builder.Build()
 				ipldGSM2 := bindnode.Unwrap(node2).(*ipldbind.GraphSyncMessage)
-				gsm2, err := messageFromIPLD(ipldGSM2)
+				gsm2, err := NewMessageHandler().messageFromIPLD(ipldGSM2)
 				require.NoError(b, err)
 
 				// same as above.

message/ipldbind/message.go

@@ -50,7 +50,7 @@ type GraphSyncRequest struct {
 	Id []byte
 
 	Root       cid.Cid
-	Selector   ipld.Node
+	Selector   *ipld.Node
 	Extensions GraphSyncExtensions
 	Priority   graphsync.Priority
 	Cancel     bool

message/ipldbind/schema.ipldsch

@@ -37,7 +37,7 @@ type GraphSyncResponseStatusCode enum {
 type GraphSyncRequest struct {
   id          GraphSyncRequestID  (rename "ID")   # unique id set on the requester side
   root        Link                (rename "Root") # a CID for the root node in the query
-  selector    Any                 (rename "Sel")  # see https://github.com/ipld/specs/blob/master/selectors/selectors.md
+  selector nullable Any                 (rename "Sel")  # see https://github.com/ipld/specs/blob/master/selectors/selectors.md
   extensions  GraphSyncExtensions (rename "Ext")  # side channel information
   priority    GraphSyncPriority   (rename "Pri")  # the priority (normalized). default to 1
   cancel      Bool                (rename "Canc") # whether this cancels a request

message/message.go

@@ -3,7 +3,6 @@ package message
 import (
 	"bytes"
 	"fmt"
-	"io"
 	"strings"
 
 	blocks "github.com/ipfs/go-block-format"
@@ -12,37 +11,8 @@ import (
 	"github.com/ipld/go-ipld-prime/codec/dagjson"
 
 	"github.com/ipfs/go-graphsync"
-	"github.com/ipfs/go-graphsync/message/ipldbind"
-	pb "github.com/ipfs/go-graphsync/message/pb"
 )
 
-// IsTerminalSuccessCode returns true if the response code indicates the
-// request terminated successfully.
-// DEPRECATED: use status.IsSuccess()
-func IsTerminalSuccessCode(status graphsync.ResponseStatusCode) bool {
-	return status.IsSuccess()
-}
-
-// IsTerminalFailureCode returns true if the response code indicates the
-// request terminated in failure.
-// DEPRECATED: use status.IsFailure()
-func IsTerminalFailureCode(status graphsync.ResponseStatusCode) bool {
-	return status.IsFailure()
-}
-
-// IsTerminalResponseCode returns true if the response code signals
-// the end of the request
-// DEPRECATED: use status.IsTerminal()
-func IsTerminalResponseCode(status graphsync.ResponseStatusCode) bool {
-	return status.IsTerminal()
-}
-
-// Exportable is an interface that can serialize to a protobuf
-type Exportable interface {
-	ToProto() (*pb.Message, error)
-	ToNet(w io.Writer) error
-}
-
 // GraphSyncRequest is a struct to capture data on a request contained in a
 // GraphSyncMessage.
 type GraphSyncRequest struct {
@@ -57,11 +27,15 @@ type GraphSyncRequest struct {
 
 // String returns a human-readable form of a GraphSyncRequest
 func (gsr GraphSyncRequest) String() string {
-	var buf bytes.Buffer
-	dagjson.Encode(gsr.selector, &buf)
+	sel := "nil"
+	if gsr.selector != nil {
+		var buf bytes.Buffer
+		dagjson.Encode(gsr.selector, &buf)
+		sel = buf.String()
+	}
 	return fmt.Sprintf("GraphSyncRequest<root=%s, selector=%s, priority=%d, id=%s, cancel=%v, update=%v, exts=%s>",
 		gsr.root.String(),
-		buf.String(),
+		sel,
 		gsr.priority,
 		gsr.id.String(),
 		gsr.isCancel,
@@ -99,7 +73,8 @@ type GraphSyncMessage struct {
 // its contents
 func (gsm GraphSyncMessage) String() string {
 	cts := make([]string, 0)
-	for _, req := range gsm.requests {
+	for i, req := range gsm.requests {
+		fmt.Printf("req.String(%v)\n", i)
 		cts = append(cts, req.String())
 	}
 	for _, resp := range gsm.responses {
@@ -213,87 +188,6 @@ func (gsm GraphSyncMessage) Blocks() []blocks.Block {
 	return bs
 }
 
-func (gsm GraphSyncMessage) ToIPLD() (*ipldbind.GraphSyncMessage, error) {
-	ibm := new(ipldbind.GraphSyncMessage)
-	ibm.Requests = make([]ipldbind.GraphSyncRequest, 0, len(gsm.requests))
-	for _, request := range gsm.requests {
-		ibm.Requests = append(ibm.Requests, ipldbind.GraphSyncRequest{
-			Id:       request.id.Bytes(),
-			Root:     request.root,
-			Selector: request.selector,
-			Priority: request.priority,
-			Cancel:   request.isCancel,
-			Update:   request.isUpdate,
-			// Extensions: request.extensions,
-		})
-	}
-
-	ibm.Responses = make([]ipldbind.GraphSyncResponse, 0, len(gsm.responses))
-	for _, response := range gsm.responses {
-		ibm.Responses = append(ibm.Responses, ipldbind.GraphSyncResponse{
-			Id:     response.requestID.Bytes(),
-			Status: response.status,
-			// Extensions: response.extensions,
-		})
-	}
-
-	blocks := gsm.Blocks()
-	ibm.Blocks = make([]ipldbind.GraphSyncBlock, 0, len(blocks))
-	for _, b := range blocks {
-		ibm.Blocks = append(ibm.Blocks, ipldbind.GraphSyncBlock{
-			Data:   b.RawData(),
-			Prefix: b.Cid().Prefix().Bytes(),
-		})
-	}
-	return ibm, nil
-}
-
-func messageFromIPLD(ibm *ipldbind.GraphSyncMessage) (GraphSyncMessage, error) {
-	requests := make(map[graphsync.RequestID]GraphSyncRequest, len(ibm.Requests))
-	for _, req := range ibm.Requests {
-		// exts := req.Extensions
-		id, err := graphsync.ParseRequestID(req.Id)
-		if err != nil {
-			return GraphSyncMessage{}, err
-		}
-		requests[id] = newRequest(id, req.Root, req.Selector, graphsync.Priority(req.Priority), req.Cancel, req.Update, nil)
-	}
-
-	responses := make(map[graphsync.RequestID]GraphSyncResponse, len(ibm.Responses))
-	for _, res := range ibm.Responses {
-		// exts := res.Extensions
-		id, err := graphsync.ParseRequestID(res.Id)
-		if err != nil {
-			return GraphSyncMessage{}, err
-		}
-		responses[id] = newResponse(id, graphsync.ResponseStatusCode(res.Status), nil)
-	}
-
-	blks := make(map[cid.Cid]blocks.Block, len(ibm.Blocks))
-	for _, b := range ibm.Blocks {
-		pref, err := cid.PrefixFromBytes(b.Prefix)
-		if err != nil {
-			return GraphSyncMessage{}, err
-		}
-
-		c, err := pref.Sum(b.Data)
-		if err != nil {
-			return GraphSyncMessage{}, err
-		}
-
-		blk, err := blocks.NewBlockWithCid(b.Data, c)
-		if err != nil {
-			return GraphSyncMessage{}, err
-		}
-
-		blks[blk.Cid()] = blk
-	}
-
-	return GraphSyncMessage{
-		requests, responses, blks,
-	}, nil
-}
-
 func (gsm GraphSyncMessage) Loggable() map[string]interface{} {
 	requests := make([]string, 0, len(gsm.requests))
 	for _, request := range gsm.requests {

message/message_test.go

@@ -45,7 +45,7 @@ func TestAppendingRequests(t *testing.T) {
 	require.True(t, found)
 	require.Equal(t, extension.Data, extensionData)
 
-	pbMessage, err := NewMessageHandler().ToProto(gsm)
+	pbMessage, err := NewMessageHandler().ToProtoV11(gsm)
 	require.NoError(t, err, "serialize to protobuf errored")
 	selectorEncoded, err := ipldutil.EncodeNode(selector)
 	require.NoError(t, err)
@@ -59,7 +59,7 @@ func TestAppendingRequests(t *testing.T) {
 	require.Equal(t, selectorEncoded, pbRequest.Selector)
 	require.Equal(t, map[string][]byte{"graphsync/awesome": extension.Data}, pbRequest.Extensions)
 
-	deserialized, err := NewMessageHandler().newMessageFromProto(pbMessage)
+	deserialized, err := NewMessageHandler().newMessageFromProtoV11(pbMessage)
 	require.NoError(t, err, "deserializing protobuf message errored")
 	deserializedRequests := deserialized.Requests()
 	require.Len(t, deserializedRequests, 1, "did not add request to deserialized message")
@@ -99,14 +99,14 @@ func TestAppendingResponses(t *testing.T) {
 	require.True(t, found)
 	require.Equal(t, extension.Data, extensionData)
 
-	pbMessage, err := NewMessageHandler().ToProto(gsm)
+	pbMessage, err := NewMessageHandler().ToProtoV11(gsm)
 	require.NoError(t, err, "serialize to protobuf errored")
 	pbResponse := pbMessage.Responses[0]
 	require.Equal(t, requestID.Bytes(), pbResponse.Id)
 	require.Equal(t, int32(status), pbResponse.Status)
 	require.Equal(t, extension.Data, pbResponse.Extensions["graphsync/awesome"])
 
-	deserialized, err := NewMessageHandler().newMessageFromProto(pbMessage)
+	deserialized, err := NewMessageHandler().newMessageFromProtoV11(pbMessage)
 	require.NoError(t, err, "deserializing protobuf message errored")
 	deserializedResponses := deserialized.Responses()
 	require.Len(t, deserializedResponses, 1, "did not add response to deserialized message")
@@ -132,7 +132,7 @@ func TestAppendBlock(t *testing.T) {
 	m, err := builder.Build()
 	require.NoError(t, err)
 
-	pbMessage, err := NewMessageHandler().ToProto(m)
+	pbMessage, err := NewMessageHandler().ToProtoV11(m)
 	require.NoError(t, err, "serializing to protobuf errored")
 
 	// assert strings are in proto message