diff --git a/go.mod b/go.mod index 509dfd0..c339321 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,9 @@ module github.com/libp2p/go-msgio -require github.com/libp2p/go-buffer-pool v0.0.2 +go 1.14 + +require ( + github.com/gogo/protobuf v1.3.1 + github.com/libp2p/go-buffer-pool v0.0.2 + github.com/multiformats/go-varint v0.0.6 +) diff --git a/go.sum b/go.sum index 23f8026..89aa941 100644 --- a/go.sum +++ b/go.sum @@ -1,4 +1,9 @@ -github.com/libp2p/go-buffer-pool v0.0.1 h1:9Rrn/H46cXjaA2HQ5Y8lyhOS1NhTkZ4yuEs2r3Eechg= -github.com/libp2p/go-buffer-pool v0.0.1/go.mod h1:xtyIz9PMobb13WaxR6Zo1Pd1zXJKYg0a8KiIvDp3TzQ= +github.com/gogo/protobuf v1.3.1 h1:DqDEcV5aeaTmdFBePNpYsp3FlcVH/2ISVVM9Qf8PSls= +github.com/gogo/protobuf v1.3.1/go.mod h1:SlYgWuQ5SjCEi6WLHjHCa1yvBfUnHcTbrrZtXPKa29o= +github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00= +github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= github.com/libp2p/go-buffer-pool v0.0.2 h1:QNK2iAFa8gjAe1SPz6mHSMuCcjs+X1wlHzeOSqcmlfs= github.com/libp2p/go-buffer-pool v0.0.2/go.mod h1:MvaB6xw5vOrDl8rYZGLFdKAuk/hRoRZd1Vi32+RXyFM= +github.com/multiformats/go-varint v0.0.6 h1:gk85QWKxh3TazbLxED/NlDVv8+q+ReFJk7Y2W/KhfNY= +github.com/multiformats/go-varint v0.0.6/go.mod h1:3Ls8CIEsrijN6+B7PbrXRPxHRPuXSrVKRY101jdMZYE= +golang.org/x/tools v0.0.0-20181030221726-6c7e314b6563/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= diff --git a/protoio/interfaces.go b/protoio/interfaces.go new file mode 100644 index 0000000..1a42713 --- /dev/null +++ b/protoio/interfaces.go @@ -0,0 +1,71 @@ +// +// Adapted from gogo/protobuf to use multiformats/go-varint for +// efficient, interoperable length-prefixing. +// +// Protocol Buffers for Go with Gadgets +// +// Copyright (c) 2013, The GoGo Authors. All rights reserved. +// http://github.com/gogo/protobuf +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +// +package protoio + +import ( + "io" + + "github.com/gogo/protobuf/proto" +) + +type Writer interface { + WriteMsg(proto.Message) error +} + +type WriteCloser interface { + Writer + io.Closer +} + +type Reader interface { + ReadMsg(msg proto.Message) error +} + +type ReadCloser interface { + Reader + io.Closer +} + +func getSize(v interface{}) (int, bool) { + if sz, ok := v.(interface { + Size() (n int) + }); ok { + return sz.Size(), true + } else if sz, ok := v.(interface { + ProtoSize() (n int) + }); ok { + return sz.ProtoSize(), true + } else { + return 0, false + } +} diff --git a/protoio/uvarint_reader.go b/protoio/uvarint_reader.go new file mode 100644 index 0000000..a3607fd --- /dev/null +++ b/protoio/uvarint_reader.go @@ -0,0 +1,83 @@ +// +// Adapted from gogo/protobuf to use multiformats/go-varint for +// efficient, interoperable length-prefixing. +// +// Protocol Buffers for Go with Gadgets +// +// Copyright (c) 2013, The GoGo Authors. All rights reserved. +// http://github.com/gogo/protobuf +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +// +package protoio + +import ( + "bufio" + "io" + + "github.com/gogo/protobuf/proto" + + "github.com/multiformats/go-varint" +) + +type uvarintReader struct { + r *bufio.Reader + buf []byte + maxSize int + closer io.Closer +} + +func NewDelimitedReader(r io.Reader, maxSize int) ReadCloser { + var closer io.Closer + if c, ok := r.(io.Closer); ok { + closer = c + } + return &uvarintReader{bufio.NewReader(r), nil, maxSize, closer} +} + +func (ur *uvarintReader) ReadMsg(msg proto.Message) error { + length64, err := varint.ReadUvarint(ur.r) + if err != nil { + return err + } + length := int(length64) + if length < 0 || length > ur.maxSize { + return io.ErrShortBuffer + } + if len(ur.buf) < length { + ur.buf = make([]byte, length) + } + buf := ur.buf[:length] + if _, err := io.ReadFull(ur.r, buf); err != nil { + return err + } + return proto.Unmarshal(buf, msg) +} + +func (ur *uvarintReader) Close() error { + if ur.closer != nil { + return ur.closer.Close() + } + return nil +} diff --git a/protoio/uvarint_test.go b/protoio/uvarint_test.go new file mode 100644 index 0000000..0cde5b2 --- /dev/null +++ b/protoio/uvarint_test.go @@ -0,0 +1,150 @@ +// +// Adapted from gogo/protobuf to use multiformats/go-varint for +// efficient, interoperable length-prefixing. +// +// Protocol Buffers for Go with Gadgets +// +// Copyright (c) 2013, The GoGo Authors. All rights reserved. +// http://github.com/gogo/protobuf +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +// +package protoio_test + +import ( + "bytes" + "io" + "math/rand" + "testing" + "time" + + "github.com/gogo/protobuf/test" + + "github.com/libp2p/go-msgio/protoio" + "github.com/multiformats/go-varint" +) + +func TestVarintNormal(t *testing.T) { + buf := newBuffer() + writer := protoio.NewDelimitedWriter(buf) + reader := protoio.NewDelimitedReader(buf, 1024*1024) + if err := iotest(writer, reader); err != nil { + t.Error(err) + } + if !buf.closed { + t.Fatalf("did not close buffer") + } +} + +func TestVarintNoClose(t *testing.T) { + buf := bytes.NewBuffer(nil) + writer := protoio.NewDelimitedWriter(buf) + reader := protoio.NewDelimitedReader(buf, 1024*1024) + if err := iotest(writer, reader); err != nil { + t.Error(err) + } +} + +// https://github.com/gogo/protobuf/issues/32 +func TestVarintMaxSize(t *testing.T) { + buf := newBuffer() + writer := protoio.NewDelimitedWriter(buf) + reader := protoio.NewDelimitedReader(buf, 20) + if err := iotest(writer, reader); err != io.ErrShortBuffer { + t.Error(err) + } else { + t.Logf("%s", err) + } +} + +func TestVarintError(t *testing.T) { + buf := newBuffer() + // beyond uvarint63 capacity. + buf.Write([]byte{0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff}) + reader := protoio.NewDelimitedReader(buf, 1024*1024) + msg := &test.NinOptNative{} + err := reader.ReadMsg(msg) + if err != varint.ErrOverflow { + t.Fatalf("expected varint.ErrOverflow error") + } +} + +type buffer struct { + *bytes.Buffer + closed bool +} + +func (b *buffer) Close() error { + b.closed = true + return nil +} + +func newBuffer() *buffer { + return &buffer{bytes.NewBuffer(nil), false} +} + +func iotest(writer protoio.WriteCloser, reader protoio.ReadCloser) error { + size := 1000 + msgs := make([]*test.NinOptNative, size) + r := rand.New(rand.NewSource(time.Now().UnixNano())) + for i := range msgs { + msgs[i] = test.NewPopulatedNinOptNative(r, true) + // https://github.com/gogo/protobuf/issues/31 + if i == 5 { + msgs[i] = &test.NinOptNative{} + } + // https://github.com/gogo/protobuf/issues/31 + if i == 999 { + msgs[i] = &test.NinOptNative{} + } + err := writer.WriteMsg(msgs[i]) + if err != nil { + return err + } + } + if err := writer.Close(); err != nil { + return err + } + i := 0 + for { + msg := &test.NinOptNative{} + if err := reader.ReadMsg(msg); err != nil { + if err == io.EOF { + break + } + return err + } + if err := msg.VerboseEqual(msgs[i]); err != nil { + return err + } + i++ + } + if i != size { + panic("not enough messages read") + } + if err := reader.Close(); err != nil { + return err + } + return nil +} diff --git a/protoio/uvarint_writer.go b/protoio/uvarint_writer.go new file mode 100644 index 0000000..0da567a --- /dev/null +++ b/protoio/uvarint_writer.go @@ -0,0 +1,93 @@ +// +// Adapted from gogo/protobuf to use multiformats/go-varint for +// efficient, interoperable length-prefixing. +// +// Protocol Buffers for Go with Gadgets +// +// Copyright (c) 2013, The GoGo Authors. All rights reserved. +// http://github.com/gogo/protobuf +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +// +package protoio + +import ( + "io" + + "github.com/gogo/protobuf/proto" + + "github.com/multiformats/go-varint" +) + +type uvarintWriter struct { + w io.Writer + lenBuf []byte + buffer []byte +} + +func NewDelimitedWriter(w io.Writer) WriteCloser { + return &uvarintWriter{w, make([]byte, varint.MaxLenUvarint63), nil} +} + +func (uw *uvarintWriter) WriteMsg(msg proto.Message) (err error) { + var data []byte + if m, ok := msg.(interface { + MarshalTo(data []byte) (n int, err error) + }); ok { + n, ok := getSize(m) + if ok { + if n+varint.MaxLenUvarint63 >= len(uw.buffer) { + uw.buffer = make([]byte, n+varint.MaxLenUvarint63) + } + lenOff := varint.PutUvarint(uw.buffer, uint64(n)) + _, err = m.MarshalTo(uw.buffer[lenOff:]) + if err != nil { + return err + } + _, err = uw.w.Write(uw.buffer[:lenOff+n]) + return err + } + } + + // fallback + data, err = proto.Marshal(msg) + if err != nil { + return err + } + length := uint64(len(data)) + n := varint.PutUvarint(uw.lenBuf, length) + _, err = uw.w.Write(uw.lenBuf[:n]) + if err != nil { + return err + } + _, err = uw.w.Write(data) + return err +} + +func (uw *uvarintWriter) Close() error { + if closer, ok := uw.w.(io.Closer); ok { + return closer.Close() + } + return nil +}