Update spdy implementation to use provider interface

Define and use the simple provider interface instead of relying on spdystream directly.

Signed-off-by: Derek McGowan <derek@mcgstyle.net> (github: dmcgowan)

Author: dmcgowan

spdy/encode.go

@@ -114,7 +114,7 @@ func (c *channel) encodeExtension(iv reflect.Value) (int, []byte, error) {
 			return 0, nil, errors.New("bad type")
 		}
 		if v.session != c.session {
-			streamCopy, err := c.session.createByteStream()
+			streamCopy, err := c.createByteStream()
 			if err != nil {
 				return 0, nil, err
 			}
@@ -133,7 +133,7 @@ func (c *channel) encodeExtension(iv reflect.Value) (int, []byte, error) {
 		return 2, b, err
 	case io.Reader:
 		// Either ReadWriteCloser, ReadWriter, or ReadCloser
-		streamCopy, err := c.session.createByteStream()
+		streamCopy, err := c.createByteStream()
 		if err != nil {
 			return 0, nil, err
 		}
@@ -154,7 +154,7 @@ func (c *channel) encodeExtension(iv reflect.Value) (int, []byte, error) {
 		b, err := streamCopy.(*byteStream).streamBytes()
 		return 2, b, err
 	case io.Writer:
-		streamCopy, err := c.session.createByteStream()
+		streamCopy, err := c.createByteStream()
 		if err != nil {
 			return 0, nil, err
 		}
@@ -208,13 +208,13 @@ func (c *channel) encodeExtension(iv reflect.Value) (int, []byte, error) {
 	return 0, nil, nil
 }
 
-func (s *Transport) decodeStream(v reflect.Value, b []byte) error {
+func (c *channel) decodeStream(v reflect.Value, b []byte) error {
 	referenceID, err := decodeReferenceID(b)
 	if err != nil {
 		return err
 	}
 
-	bs := s.getByteStream(referenceID)
+	bs := c.session.getByteStream(referenceID)
 	if bs != nil {
 		v.Set(reflect.ValueOf(bs))
 	}
@@ -226,6 +226,6 @@ func (c *channel) initializeExtensions() *msgpack.Extensions {
 	exts := msgpack.NewExtensions()
 	exts.SetEncoder(c.encodeExtension)
 	exts.AddDecoder(1, reflect.TypeOf(&channel{}), c.decodeChannel)
-	exts.AddDecoder(2, reflect.TypeOf(&byteStream{}), c.session.decodeStream)
+	exts.AddDecoder(2, reflect.TypeOf(&byteStream{}), c.decodeStream)
 	return exts
 }

spdy/listener.go

@@ -1,45 +1,48 @@
 package spdy
 
 import (
+	"io"
 	"net"
 
 	"github.com/docker/libchan"
 )
 
 type pipeSender struct {
-	session *Transport
+	session libchan.Transport
 	sender  *channel
 }
 
 type pipeReceiver struct {
-	session  *Transport
+	session  libchan.Transport
 	receiver *channel
 }
 
 // Pipe creates a top-level channel pipe using an in memory transport.
 func Pipe() (libchan.Receiver, libchan.Sender, error) {
 	c1, c2 := net.Pipe()
 
-	s1, err := newSession(c1, false)
+	s1, err := NewSpdyStreamProvider(c1, false)
 	if err != nil {
 		return nil, nil, err
 	}
+	t1 := NewTransport(s1)
 
-	s2, err := newSession(c2, true)
+	s2, err := NewSpdyStreamProvider(c2, true)
 	if err != nil {
 		return nil, nil, err
 	}
+	t2 := NewTransport(s2)
 
 	var receiver libchan.Receiver
 	waitError := make(chan error)
 
 	go func() {
 		var err error
-		receiver, err = s2.WaitReceiveChannel()
+		receiver, err = t2.WaitReceiveChannel()
 		waitError <- err
 	}()
 
-	sender, senderErr := s1.NewSendChannel()
+	sender, senderErr := t1.NewSendChannel()
 	if senderErr != nil {
 		c1.Close()
 		c2.Close()
@@ -52,7 +55,7 @@ func Pipe() (libchan.Receiver, libchan.Sender, error) {
 		c2.Close()
 		return nil, nil, receiveErr
 	}
-	return &pipeReceiver{s2, receiver.(*channel)}, &pipeSender{s1, sender.(*channel)}, nil
+	return &pipeReceiver{t2, receiver.(*channel)}, &pipeSender{t1, sender.(*channel)}, nil
 }
 
 func (p *pipeSender) Send(message interface{}) error {
@@ -64,7 +67,10 @@ func (p *pipeSender) Close() error {
 	if err != nil {
 		return err
 	}
-	return p.session.Close()
+	if closer, ok := p.session.(io.Closer); ok {
+		return closer.Close()
+	}
+	return nil
 }
 
 func (p *pipeReceiver) Receive(message interface{}) error {