fix(net): refactor message version format code to separate packages

Author: rvagg

benchmarks/testnet/virtual.go

@@ -16,6 +16,7 @@ import (
 	"google.golang.org/protobuf/proto"
 
 	gsmsg "github.com/ipfs/go-graphsync/message"
+	gsmsgv1 "github.com/ipfs/go-graphsync/message/v1"
 	gsnet "github.com/ipfs/go-graphsync/network"
 )
 
@@ -137,7 +138,7 @@ func (n *network) SendMessage(
 			rateLimiters[to] = rateLimiter
 		}
 
-		pbMsg, err := gsmsg.NewMessageHandler().ToProtoV11(mes)
+		pbMsg, err := gsmsgv1.NewMessageHandler().ToProto(peer.ID("foo"), mes)
 		if err != nil {
 			return err
 		}

message/bench_test.go renamed to message/bench/bench_test.go

@@ -8,14 +8,16 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	blocks "github.com/ipfs/go-block-format"
-	"github.com/ipld/go-ipld-prime/codec/dagcbor"
+	"github.com/ipld/go-ipld-prime"
 	"github.com/ipld/go-ipld-prime/node/basicnode"
-	"github.com/ipld/go-ipld-prime/node/bindnode"
 	"github.com/ipld/go-ipld-prime/traversal/selector/builder"
+	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/stretchr/testify/require"
 
 	"github.com/ipfs/go-graphsync"
-	"github.com/ipfs/go-graphsync/message/ipldbind"
+	"github.com/ipfs/go-graphsync/message"
+	v1 "github.com/ipfs/go-graphsync/message/v1"
+	v2 "github.com/ipfs/go-graphsync/message/v2"
 	"github.com/ipfs/go-graphsync/testutil"
 )
 
@@ -34,35 +36,40 @@ func BenchmarkMessageEncodingRoundtrip(b *testing.B) {
 	priority := graphsync.Priority(rand.Int31())
 	status := graphsync.RequestAcknowledged
 
-	builder := NewBuilder()
-	builder.AddRequest(NewRequest(id, root, selector, priority, extension))
-	builder.AddRequest(NewRequest(id, root, selector, priority))
+	builder := message.NewBuilder()
+	builder.AddRequest(message.NewRequest(id, root, selector, priority, extension))
+	builder.AddRequest(message.NewRequest(id, root, selector, priority))
 	builder.AddResponseCode(id, status)
 	builder.AddExtensionData(id, extension)
 	builder.AddBlock(blocks.NewBlock([]byte("W")))
 	builder.AddBlock(blocks.NewBlock([]byte("E")))
 	builder.AddBlock(blocks.NewBlock([]byte("F")))
 	builder.AddBlock(blocks.NewBlock([]byte("M")))
 
+	p := peer.ID("test peer")
 	gsm, err := builder.Build()
 	require.NoError(b, err)
 
 	b.Run("Protobuf", func(b *testing.B) {
 		b.ReportAllocs()
 		b.RunParallel(func(pb *testing.PB) {
 			buf := new(bytes.Buffer)
+			mh := v1.NewMessageHandler()
 			for pb.Next() {
 				buf.Reset()
 
-				err := NewMessageHandler().ToNet(gsm, buf)
+				err := mh.ToNet(p, gsm, buf)
 				require.NoError(b, err)
 
-				gsm2, err := NewMessageHandler().FromNet(buf)
+				gsm2, err := mh.FromNet(p, buf)
 				require.NoError(b, err)
 
 				// Note that require.Equal doesn't seem to handle maps well.
 				// It says they are non-equal simply because their order isn't deterministic.
-				if diff := cmp.Diff(gsm, gsm2, cmp.Exporter(func(reflect.Type) bool { return true })); diff != "" {
+				if diff := cmp.Diff(gsm, gsm2,
+					cmp.Exporter(func(reflect.Type) bool { return true }),
+					cmp.Comparer(ipld.DeepEqual),
+				); diff != "" {
 					b.Fatal(diff)
 				}
 			}
@@ -73,25 +80,21 @@ func BenchmarkMessageEncodingRoundtrip(b *testing.B) {
 		b.ReportAllocs()
 		b.RunParallel(func(pb *testing.PB) {
 			buf := new(bytes.Buffer)
+			mh := v2.NewMessageHandler()
 			for pb.Next() {
 				buf.Reset()
 
-				ipldGSM, err := NewMessageHandler().toIPLD(gsm)
-				require.NoError(b, err)
-				node := bindnode.Wrap(ipldGSM, ipldbind.Prototype.Message.Type())
-				err = dagcbor.Encode(node.Representation(), buf)
+				err := mh.ToNet(p, gsm, buf)
 				require.NoError(b, err)
 
-				builder := ipldbind.Prototype.Message.Representation().NewBuilder()
-				err = dagcbor.Decode(builder, buf)
-				require.NoError(b, err)
-				node2 := builder.Build()
-				ipldGSM2 := bindnode.Unwrap(node2).(*ipldbind.GraphSyncMessage)
-				gsm2, err := NewMessageHandler().fromIPLD(ipldGSM2)
+				gsm2, err := mh.FromNet(p, buf)
 				require.NoError(b, err)
 
 				// same as above.
-				if diff := cmp.Diff(gsm, gsm2, cmp.Exporter(func(reflect.Type) bool { return true })); diff != "" {
+				if diff := cmp.Diff(gsm, gsm2,
+					cmp.Exporter(func(reflect.Type) bool { return true }),
+					cmp.Comparer(ipld.DeepEqual),
+				); diff != "" {
 					b.Fatal(diff)
 				}
 			}

message/ipldbind/message.go

@@ -1,56 +1,42 @@
 package ipldbind
 
 import (
-	"io"
-
-	blocks "github.com/ipfs/go-block-format"
 	cid "github.com/ipfs/go-cid"
 	"github.com/ipld/go-ipld-prime/datamodel"
 
 	"github.com/ipfs/go-graphsync"
-	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
+type MessagePartWithExtensions interface {
+	ExtensionNames() []graphsync.ExtensionName
+	Extension(name graphsync.ExtensionName) (datamodel.Node, bool)
 }
 
 type GraphSyncExtensions struct {
 	Keys   []string
 	Values map[string]datamodel.Node
 }
 
-func NewGraphSyncExtensions(values map[string]datamodel.Node) GraphSyncExtensions {
-	keys := make([]string, 0, len(values))
-	for k := range values {
-		keys = append(keys, k)
+func NewGraphSyncExtensions(part MessagePartWithExtensions) GraphSyncExtensions {
+	names := part.ExtensionNames()
+	keys := make([]string, 0, len(names))
+	values := make(map[string]datamodel.Node, len(names))
+	for _, name := range names {
+		keys = append(keys, string(name))
+		data, _ := part.Extension(graphsync.ExtensionName(name))
+		values[string(name)] = data
 	}
 	return GraphSyncExtensions{keys, values}
 }
 
+func (gse GraphSyncExtensions) ToExtensionsList() []graphsync.ExtensionData {
+	exts := make([]graphsync.ExtensionData, 0, len(gse.Values))
+	for name, data := range gse.Values {
+		exts = append(exts, graphsync.ExtensionData{Name: graphsync.ExtensionName(name), Data: data})
+	}
+	return exts
+}
+
 // GraphSyncRequest is a struct to capture data on a request contained in a
 // GraphSyncMessage.
 type GraphSyncRequest struct {
@@ -84,39 +70,6 @@ type GraphSyncBlock struct {
 	Data   []byte
 }
 
-func FromBlockFormat(block blocks.Block) GraphSyncBlock {
-	return GraphSyncBlock{
-		Prefix: block.Cid().Prefix().Bytes(),
-		Data:   block.RawData(),
-	}
-}
-
-func (b GraphSyncBlock) BlockFormat() *blocks.BasicBlock {
-	pref, err := cid.PrefixFromBytes(b.Prefix)
-	if err != nil {
-		panic(err) // should never happen
-	}
-
-	c, err := pref.Sum(b.Data)
-	if err != nil {
-		panic(err) // should never happen
-	}
-
-	block, err := blocks.NewBlockWithCid(b.Data, c)
-	if err != nil {
-		panic(err) // should never happen
-	}
-	return block
-}
-
-func BlockFormatSlice(bs []GraphSyncBlock) []blocks.Block {
-	blks := make([]blocks.Block, len(bs))
-	for i, b := range bs {
-		blks[i] = b.BlockFormat()
-	}
-	return blks
-}
-
 type GraphSyncMessage struct {
 	Requests  []GraphSyncRequest
 	Responses []GraphSyncResponse

message/ipldbind/schema.go

@@ -11,8 +11,6 @@ import (
 //go:embed schema.ipldsch
 var embedSchema []byte
 
-var schemaTypeSystem *schema.TypeSystem
-
 var Prototype struct {
 	Message schema.TypedPrototype
 }
@@ -22,7 +20,6 @@ func init() {
 	if err != nil {
 		panic(err)
 	}
-	schemaTypeSystem = ts
 
 	Prototype.Message = bindnode.Prototype((*GraphSyncMessage)(nil), ts.TypeByName("GraphSyncMessage"))
 }

message/message.go

@@ -3,17 +3,26 @@ package message
 import (
 	"bytes"
 	"fmt"
+	"io"
 	"strings"
 
 	blocks "github.com/ipfs/go-block-format"
 	cid "github.com/ipfs/go-cid"
 	"github.com/ipld/go-ipld-prime"
 	"github.com/ipld/go-ipld-prime/codec/dagjson"
 	"github.com/ipld/go-ipld-prime/datamodel"
+	"github.com/libp2p/go-libp2p-core/peer"
+	"github.com/libp2p/go-msgio"
 
 	"github.com/ipfs/go-graphsync"
 )
 
+type MessageHandler interface {
+	FromNet(peer.ID, io.Reader) (GraphSyncMessage, error)
+	FromMsgReader(peer.ID, msgio.Reader) (GraphSyncMessage, error)
+	ToNet(peer.ID, GraphSyncMessage, io.Writer) error
+}
+
 // GraphSyncRequest is a struct to capture data on a request contained in a
 // GraphSyncMessage.
 type GraphSyncRequest struct {
@@ -34,14 +43,19 @@ func (gsr GraphSyncRequest) String() string {
 		dagjson.Encode(gsr.selector, &buf)
 		sel = buf.String()
 	}
+	extStr := strings.Builder{}
+	for _, name := range gsr.ExtensionNames() {
+		extStr.WriteString(string(name))
+		extStr.WriteString("|")
+	}
 	return fmt.Sprintf("GraphSyncRequest<root=%s, selector=%s, priority=%d, id=%s, cancel=%v, update=%v, exts=%s>",
 		gsr.root.String(),
 		sel,
 		gsr.priority,
 		gsr.id.String(),
 		gsr.isCancel,
 		gsr.isUpdate,
-		strings.Join(gsr.ExtensionNames(), "|"),
+		extStr.String(),
 	)
 }
 
@@ -55,10 +69,15 @@ type GraphSyncResponse struct {
 
 // String returns a human-readable form of a GraphSyncResponse
 func (gsr GraphSyncResponse) String() string {
+	extStr := strings.Builder{}
+	for _, name := range gsr.ExtensionNames() {
+		extStr.WriteString(string(name))
+		extStr.WriteString("|")
+	}
 	return fmt.Sprintf("GraphSyncResponse<id=%s, status=%d, exts=%s>",
 		gsr.requestID.String(),
 		gsr.status,
-		strings.Join(gsr.ExtensionNames(), "|"),
+		extStr.String(),
 	)
 }
 
@@ -70,6 +89,14 @@ type GraphSyncMessage struct {
 	blocks    map[cid.Cid]blocks.Block
 }
 
+func NewMessage(
+	requests map[graphsync.RequestID]GraphSyncRequest,
+	responses map[graphsync.RequestID]GraphSyncResponse,
+	blocks map[cid.Cid]blocks.Block,
+) GraphSyncMessage {
+	return GraphSyncMessage{requests, responses, blocks}
+}
+
 // String returns a human-readable (multi-line) form of a GraphSyncMessage and
 // its contents
 func (gsm GraphSyncMessage) String() string {
@@ -87,7 +114,7 @@ func (gsm GraphSyncMessage) String() string {
 	return fmt.Sprintf("GraphSyncMessage<\n\t%s\n>", strings.Join(cts, ",\n\t"))
 }
 
-// NewRequest builds a new Graphsync request
+// NewRequest builds a new GraphSyncRequest
 func NewRequest(id graphsync.RequestID,
 	root cid.Cid,
 	selector ipld.Node,
@@ -97,13 +124,13 @@ func NewRequest(id graphsync.RequestID,
 	return newRequest(id, root, selector, priority, false, false, toExtensionsMap(extensions))
 }
 
-// CancelRequest request generates a request to cancel an in progress request
-func CancelRequest(id graphsync.RequestID) GraphSyncRequest {
+// NewCancelRequest request generates a request to cancel an in progress request
+func NewCancelRequest(id graphsync.RequestID) GraphSyncRequest {
 	return newRequest(id, cid.Cid{}, nil, 0, true, false, nil)
 }
 
-// UpdateRequest generates a new request to update an in progress request with the given extensions
-func UpdateRequest(id graphsync.RequestID, extensions ...graphsync.ExtensionData) GraphSyncRequest {
+// NewUpdateRequest generates a new request to update an in progress request with the given extensions
+func NewUpdateRequest(id graphsync.RequestID, extensions ...graphsync.ExtensionData) GraphSyncRequest {
 	return newRequest(id, cid.Cid{}, nil, 0, false, true, toExtensionsMap(extensions))
 }
 
@@ -247,10 +274,10 @@ func (gsr GraphSyncRequest) Extension(name graphsync.ExtensionName) (datamodel.N
 }
 
 // ExtensionNames returns the names of the extensions included in this request
-func (gsr GraphSyncRequest) ExtensionNames() []string {
-	var extNames []string
+func (gsr GraphSyncRequest) ExtensionNames() []graphsync.ExtensionName {
+	var extNames []graphsync.ExtensionName
 	for ext := range gsr.extensions {
-		extNames = append(extNames, ext)
+		extNames = append(extNames, graphsync.ExtensionName(ext))
 	}
 	return extNames
 }
@@ -281,10 +308,10 @@ func (gsr GraphSyncResponse) Extension(name graphsync.ExtensionName) (datamodel.
 }
 
 // ExtensionNames returns the names of the extensions included in this request
-func (gsr GraphSyncResponse) ExtensionNames() []string {
-	var extNames []string
+func (gsr GraphSyncResponse) ExtensionNames() []graphsync.ExtensionName {
+	var extNames []graphsync.ExtensionName
 	for ext := range gsr.extensions {
-		extNames = append(extNames, ext)
+		extNames = append(extNames, graphsync.ExtensionName(ext))
 	}
 	return extNames
 }