Add Protocol Buffer Reader (and break existing codecs)

coding/Makefile

@@ -12,6 +12,11 @@ bin/convert:
 	mkdir -p bin
 	cd bin; go build convert.go
 
+bin/msgio:
+	mkdir -p bin
+	go get -d github.com/jbenet/go-msgio/msgio
+	go build -o "$@" github.com/jbenet/go-msgio/msgio
+
 json.testfile: bin/multicodec Makefile test1.json
 	: >$@
 	bin/multicodec header /mdagv1 >>$@
@@ -24,3 +29,11 @@ cbor.testfile: bin/multicodec bin/json2cbor Makefile test1.json
 	bin/multicodec header /cbor >>$@
 	bin/json2cbor -i test1.json -o - >>$@
 
+protobuf.testfile: bin/multicodec bin/msgio
+	bin/multicodec header /mdagv1 >$@
+	bin/multicodec header /protobuf/msgio >>$@
+	mkdir -p dir
+	echo a >dir/a
+	echo b >dir/b
+	hash=`ipfs add -q -r dir | tail -n1` && \
+		ipfs object get "$$hash" --enc=protobuf | bin/msgio wrap >>$@

coding/coding.go

@@ -8,14 +8,19 @@ import (
 	json "github.com/ipfs/go-ipld/coding/json"
 	mc "github.com/jbenet/go-multicodec"
 	mcmux "github.com/jbenet/go-multicodec/mux"
+	mcproto "github.com/jbenet/go-multicodec/protobuf"
 
 	ipld "github.com/ipfs/go-ipld"
 	pb "github.com/ipfs/go-ipld/coding/pb"
 	memory "github.com/ipfs/go-ipld/memory"
 )
 
 var Header []byte
-var HeaderPath string
+
+const (
+	HeaderPath   = "/mdagv1"
+	ProtobufPath = "/protobuf/msgio"
+)
 
 var StreamCodecs map[string]func(io.Reader) (ipld.NodeReader, error)
 
@@ -29,7 +34,6 @@ var defaultCodec string
 var muxCodec *mcmux.Multicodec
 
 func init() {
-	HeaderPath = "/mdagv1"
 	Header = mc.Header([]byte(HeaderPath))
 	// by default, always encode things as cbor
 	defaultCodec = string(mc.HeaderPath(cbor.Header))
@@ -47,6 +51,7 @@ func init() {
 		cbor.HeaderPath: func(r io.Reader) (ipld.NodeReader, error) {
 			return cbor.NewCBORDecoder(r)
 		},
+		ProtobufPath: DecodeLegacyProtobuf,
 	}
 }
 
@@ -58,46 +63,13 @@ func Multicodec() mc.Multicodec {
 }
 
 func selectCodec(v interface{}, codecs []mc.Multicodec) mc.Multicodec {
-	vn, ok := v.(*memory.Node)
-	if !ok {
-		return nil
-	}
-
-	codecKey, err := codecKey(*vn)
-	if err != nil {
-		return nil
-	}
-
-	for _, c := range codecs {
-		if codecKey == string(mc.HeaderPath(c.Header())) {
-			return c
-		}
-	}
-
 	return nil // no codec
 }
 
-func codecKey(n memory.Node) (string, error) {
-	chdr, ok := (n)[ipld.CodecKey]
-	if !ok {
-		// if no codec is defined, use our default codec
-		chdr = defaultCodec
-		if pb.IsOldProtobufNode(n) {
-			chdr = string(pb.Header)
-		}
-	}
-
-	chdrs, ok := chdr.(string)
-	if !ok {
-		// if chdr is not a string, cannot read codec.
-		return "", mc.ErrType
-	}
-
-	return chdrs, nil
-}
-
 func Decode(r io.Reader) (ipld.NodeReader, error) {
-	if err := mc.ConsumeHeader(r, mcmux.Header); err != nil {
+	// get multicodec first header, should be mcmux.Header
+	err := mc.ConsumeHeader(r, Header)
+	if err != nil {
 		return nil, err
 	}
 
@@ -111,7 +83,15 @@ func Decode(r io.Reader) (ipld.NodeReader, error) {
 
 	fun, ok := StreamCodecs[hdrPath]
 	if !ok {
-		return nil, fmt.Errorf("no codec for %s", hdr)
+		return nil, fmt.Errorf("no codec for %s", hdrPath)
 	}
 	return fun(r)
 }
+
+func DecodeLegacyProtobuf(r io.Reader) (ipld.NodeReader, error) {
+	var node memory.Node = memory.Node{}
+	r = mc.WrapHeaderReader(mcproto.HeaderMsgio, r)
+	r = mc.WrapHeaderReader(pb.Header, r)
+	err := pb.Multicodec().Decoder(r).Decode(&node)
+	return node, err
+}

coding/coding_test.go

@@ -16,8 +16,9 @@ import (
 )
 
 var codedFiles map[string][]byte = map[string][]byte{
-	"json.testfile": []byte{},
-	"cbor.testfile": []byte{},
+	"json.testfile":     []byte{},
+	"cbor.testfile":     []byte{},
+	"protobuf.testfile": []byte{},
 }
 
 func init() {
@@ -111,11 +112,13 @@ func TestRoundtripBasicMC(t *testing.T) {
 
 // Test decoding and encoding a json and cbor file
 func TestCodecsDecodeEncode(t *testing.T) {
-	for fname, testfile := range codedFiles {
+	for _, fname := range []string{"json.testfile", "cbor.testfile"} {
+		testfile := codedFiles[fname]
 		var n memory.Node
 		codec := Multicodec()
 
 		if err := mc.Unmarshal(codec, testfile, &n); err != nil {
+			t.Log(fname)
 			t.Log(testfile)
 			t.Error(err)
 			continue
@@ -188,3 +191,45 @@ func TestCborStream(t *testing.T) {
 		readertest.Callback{[]interface{}{}, reader.TokenEndNode, nil},
 	})
 }
+
+func TestPbStream(t *testing.T) {
+	a := assrt.NewAssert(t)
+	t.Logf("Reading protobuf.testfile")
+	pb, err := Decode(bytes.NewReader(codedFiles["protobuf.testfile"]))
+	a.MustNil(err)
+
+	readertest.CheckReader(t, pb, []readertest.Callback{
+		readertest.Callback{[]interface{}{}, reader.TokenNode, nil},
+		readertest.Callback{[]interface{}{}, reader.TokenKey, "data"},
+		readertest.Callback{[]interface{}{"data"}, reader.TokenValue, []byte{0x08, 0x01}},
+		readertest.Callback{[]interface{}{}, reader.TokenKey, "named-links"},
+		readertest.Callback{[]interface{}{"named-links"}, reader.TokenNode, nil},
+		readertest.Callback{[]interface{}{"named-links"}, reader.TokenKey, "a"},
+		readertest.Callback{[]interface{}{"named-links", "a"}, reader.TokenNode, nil},
+		readertest.Callback{[]interface{}{"named-links", "a"}, reader.TokenKey, "link"},
+		readertest.Callback{[]interface{}{"named-links", "a", "link"}, reader.TokenValue, "Qmbvkmk9LFsGneteXk3G7YLqtLVME566ho6ibaQZZVHaC9"},
+		readertest.Callback{[]interface{}{"named-links", "a"}, reader.TokenKey, "name"},
+		readertest.Callback{[]interface{}{"named-links", "a", "name"}, reader.TokenValue, "a"},
+		readertest.Callback{[]interface{}{"named-links", "a"}, reader.TokenKey, "size"},
+		readertest.Callback{[]interface{}{"named-links", "a", "size"}, reader.TokenValue, uint64(10)},
+		readertest.Callback{[]interface{}{"named-links", "a"}, reader.TokenEndNode, nil},
+		readertest.Callback{[]interface{}{"named-links"}, reader.TokenKey, "b"},
+		readertest.Callback{[]interface{}{"named-links", "b"}, reader.TokenNode, nil},
+		readertest.Callback{[]interface{}{"named-links", "b"}, reader.TokenKey, "link"},
+		readertest.Callback{[]interface{}{"named-links", "b", "link"}, reader.TokenValue, "QmR9pC5uCF3UExca8RSrCVL8eKv7nHMpATzbEQkAHpXmVM"},
+		readertest.Callback{[]interface{}{"named-links", "b"}, reader.TokenKey, "name"},
+		readertest.Callback{[]interface{}{"named-links", "b", "name"}, reader.TokenValue, "b"},
+		readertest.Callback{[]interface{}{"named-links", "b"}, reader.TokenKey, "size"},
+		readertest.Callback{[]interface{}{"named-links", "b", "size"}, reader.TokenValue, uint64(10)},
+		readertest.Callback{[]interface{}{"named-links", "b"}, reader.TokenEndNode, nil},
+		readertest.Callback{[]interface{}{"named-links"}, reader.TokenEndNode, nil},
+		readertest.Callback{[]interface{}{}, reader.TokenKey, "ordered-links"},
+		readertest.Callback{[]interface{}{"ordered-links"}, reader.TokenArray, nil},
+		readertest.Callback{[]interface{}{"ordered-links"}, reader.TokenIndex, 0},
+		readertest.Callback{[]interface{}{"ordered-links", 0}, reader.TokenValue, "a"},
+		readertest.Callback{[]interface{}{"ordered-links"}, reader.TokenIndex, 1},
+		readertest.Callback{[]interface{}{"ordered-links", 1}, reader.TokenValue, "b"},
+		readertest.Callback{[]interface{}{"ordered-links"}, reader.TokenEndArray, nil},
+		readertest.Callback{[]interface{}{}, reader.TokenEndNode, nil},
+	})
+}

coding/pb/Makefile

@@ -10,16 +10,14 @@ clean:
 	rm -f *.pb.go
 	rm -f *.go
 
-testfile: ../bin/multicodec bin/msgio
+testfile: ../bin/multicodec ../bin/msgio
 	../bin/multicodec header /mdagv1 >testfile
 	../bin/multicodec header /protobuf/msgio >>testfile
 	hash=`ipfs add -q -r . | tail -n1` && \
-		ipfs object get "$$hash" --enc=protobuf | bin/msgio wrap >>testfile
+		ipfs object get "$$hash" --enc=protobuf | ../bin/msgio wrap >>testfile
 
-bin/msgio:
-	mkdir -p bin
-	go get -d github.com/jbenet/go-msgio/msgio
-	go build -o "$@" github.com/jbenet/go-msgio/msgio
+../bin/msgio:
+	$(MAKE) -C .. bin/msgio
 
 ../bin/multicodec:
 	$(MAKE) -C .. bin/multicodec