Move msgpack encoding into separate package

Signed-off-by: Derek McGowan <[email protected]> (github: dmcgowan)

Author: dmcgowan

spdy/encode.go renamed to encoding/msgpack/codec.go

@@ -1,4 +1,4 @@
-package spdy
+package msgpack
 
 import (
 	"encoding/binary"
@@ -224,13 +224,13 @@ func (p *cproducer) encodeExtended(iv reflect.Value) (i int, b []byte, e error)
 	return 0, nil, nil
 }
 
-// MsgpackCodec implements the libchan encoding using msgpack5.
-type MsgpackCodec struct{}
+// Codec implements the libchan encoding using msgpack5.
+type Codec struct{}
 
 // NewEncoder returns a libchan encoder which encodes given objects
 // to msgpack5 on the given datastream using the given encoding
 // channel producer.
-func (codec *MsgpackCodec) NewEncoder(w io.Writer, p encoding.ChanProducer) encoding.Encoder {
+func (codec *Codec) NewEncoder(w io.Writer, p encoding.ChanProducer) encoding.Encoder {
 	prd := &cproducer{p}
 	encoder := msgpack.NewEncoder(w)
 	exts := msgpack.NewExtensions()
@@ -242,7 +242,7 @@ func (codec *MsgpackCodec) NewEncoder(w io.Writer, p encoding.ChanProducer) enco
 // NewDecoder returns a libchan decoder which decodes objects from
 // the given data stream from msgpack5 into provided object using
 // the provided types for libchan interfaces.
-func (codec *MsgpackCodec) NewDecoder(r io.Reader, recv encoding.ChanReceiver, streamT, recvT, sendT reflect.Type) encoding.Decoder {
+func (codec *Codec) NewDecoder(r io.Reader, recv encoding.ChanReceiver, streamT, recvT, sendT reflect.Type) encoding.Decoder {
 	rec := &creceiver{recv}
 	decoder := msgpack.NewDecoder(r)
 	exts := msgpack.NewExtensions()
@@ -259,6 +259,6 @@ func (codec *MsgpackCodec) NewDecoder(r io.Reader, recv encoding.ChanReceiver, s
 // NewRawMessage returns a transit object which will copy a
 // msgpack5 datastream and allow decoding that object
 // using a Decoder from the codec object.
-func (codec *MsgpackCodec) NewRawMessage() encoding.Decoder {
+func (codec *Codec) NewRawMessage() encoding.Decoder {
 	return new(msgpack.RawMessage)
 }

spdy/pipe.go

@@ -1,22 +1,12 @@
 package spdy
 
 import (
-	"io"
 	"net"
 
 	"github.com/docker/libchan"
+	"github.com/docker/libchan/encoding/msgpack"
 )
 
-type pipeSender struct {
-	session libchan.Transport
-	sender  *sender
-}
-
-type pipeReceiver struct {
-	session  libchan.Transport
-	receiver *receiver
-}
-
 // Pipe creates a top-level channel pipe using an in memory transport.
 func Pipe() (libchan.Receiver, libchan.Sender, error) {
 	c1, c2 := net.Pipe()
@@ -25,13 +15,13 @@ func Pipe() (libchan.Receiver, libchan.Sender, error) {
 	if err != nil {
 		return nil, nil, err
 	}
-	t1 := NewTransport(s1, &MsgpackCodec{})
+	t1 := NewTransport(s1, &msgpack.Codec{})
 
 	s2, err := NewSpdyStreamProvider(c2, true)
 	if err != nil {
 		return nil, nil, err
 	}
-	t2 := NewTransport(s2, &MsgpackCodec{})
+	t2 := NewTransport(s2, &msgpack.Codec{})
 
 	var recv libchan.Receiver
 	waitError := make(chan error)
@@ -55,28 +45,5 @@ func Pipe() (libchan.Receiver, libchan.Sender, error) {
 		c2.Close()
 		return nil, nil, receiveErr
 	}
-	return &pipeReceiver{t2, recv.(*receiver)}, &pipeSender{t1, send.(*sender)}, nil
-}
-
-func (p *pipeSender) Send(message interface{}) error {
-	return p.sender.Send(message)
-}
-
-func (p *pipeSender) Close() error {
-	err := p.sender.Close()
-	if err != nil {
-		return err
-	}
-	if closer, ok := p.session.(io.Closer); ok {
-		return closer.Close()
-	}
-	return nil
-}
-
-func (p *pipeReceiver) Receive(message interface{}) error {
-	return p.receiver.Receive(message)
-}
-
-func (p *pipeReceiver) SendTo(dst libchan.Sender) (int, error) {
-	return p.receiver.SendTo(dst)
+	return recv, send, nil
 }

spdy/session_test.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/docker/libchan"
+	"github.com/docker/libchan/encoding/msgpack"
 )
 
 type InOutMessage struct {
@@ -448,7 +449,7 @@ func ClientSendWrapper(f func(t *testing.T, c libchan.Sender, s libchan.Transpor
 		if sessionErr != nil {
 			t.Fatalf("Error creating session: %s", sessionErr)
 		}
-		session := NewTransport(provider, &MsgpackCodec{})
+		session := NewTransport(provider, &msgpack.Codec{})
 
 		sender, senderErr := session.NewSendChannel()
 		if senderErr != nil {
@@ -480,7 +481,7 @@ func ServerReceiveWrapper(f func(t *testing.T, c libchan.Receiver, s libchan.Tra
 		if sessionErr != nil {
 			t.Fatalf("Error creating session: %s", sessionErr)
 		}
-		session := NewTransport(provider, &MsgpackCodec{})
+		session := NewTransport(provider, &msgpack.Codec{})
 
 		receiver, receiverErr := session.WaitReceiveChannel()
 		if receiverErr != nil {