Reduce transport interface and expose underlying spdy transport

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

Author: dmcgowan

libchan.go

@@ -2,7 +2,6 @@ package libchan
 
 import (
 	"io"
-	"net"
 )
 
 // Transport represents a connection which can multiplex channels and
@@ -16,27 +15,6 @@ type Transport interface {
 	// WaitReceiveChannel waits for a new channel be created by the
 	// remote end of the transport calling NewSendChannel.
 	WaitReceiveChannel() (Receiver, error)
-
-	// RegisterConn registers a network connection to be used
-	// by inbound messages referring to the connection
-	// with the registered connection's local and remote address.
-	// Note: a connection does not need to be registered before
-	// being sent in a message, but does need to be registered
-	// to by the receiver of a message. If registration should be
-	// automatic, register a listener instead.
-	RegisterConn(net.Conn) error
-
-	// RegisterListener accepts all connections from the listener
-	// and immediately registers them.
-	RegisterListener(net.Listener)
-
-	// Unregister removes the connection from the list of known
-	// connections. This should be called when a connection is
-	// closed and no longer expected in inbound messages.
-	// Failure to unregister connections will increase memory
-	// usage since the transport is not notified of closed
-	// connections to automatically unregister.
-	Unregister(net.Conn)
 }
 
 // Sender is a channel which sent messages of any content

spdy/conn_test.go

@@ -25,7 +25,7 @@ type RecvMultiMessage struct {
 
 func TestMultiTcpByteStream(t *testing.T) {
 	wait := make(chan bool)
-	client := func(t *testing.T, sender libchan.Sender, s *session) {
+	client := func(t *testing.T, sender libchan.Sender, s *SpdyTransport) {
 		<-wait
 		both, connErr := net.Dial("tcp", "localhost:9272")
 		if connErr != nil {
@@ -97,7 +97,7 @@ func TestMultiTcpByteStream(t *testing.T) {
 		in.Close()
 		out.Close()
 	}
-	server := func(t *testing.T, receiver libchan.Receiver, s *session) {
+	server := func(t *testing.T, receiver libchan.Receiver, s *SpdyTransport) {
 		listener, listenerErr := net.Listen("tcp", "localhost:9272")
 		if listenerErr != nil {
 			t.Fatalf("Error creating byte stream listener: %s", listenerErr)

spdy/encode.go

@@ -10,7 +10,7 @@ import (
 	"github.com/dmcgowan/go/codec"
 )
 
-func (s *session) encodeChannel(v reflect.Value) ([]byte, error) {
+func (s *SpdyTransport) encodeChannel(v reflect.Value) ([]byte, error) {
 	rc := v.Interface().(channel)
 	if rc.stream == nil {
 		return nil, errors.New("bad type")
@@ -36,7 +36,7 @@ func (s *session) encodeChannel(v reflect.Value) ([]byte, error) {
 	return buf[:(written + 1)], nil
 }
 
-func (s *session) decodeChannel(v reflect.Value, b []byte) error {
+func (s *SpdyTransport) decodeChannel(v reflect.Value, b []byte) error {
 	rc := v.Interface().(channel)
 
 	if b[0] == 0x01 {
@@ -62,7 +62,7 @@ func (s *session) decodeChannel(v reflect.Value, b []byte) error {
 	return nil
 }
 
-func (s *session) encodeStream(v reflect.Value) ([]byte, error) {
+func (s *SpdyTransport) encodeStream(v reflect.Value) ([]byte, error) {
 	bs := v.Interface().(byteStream)
 	if bs.ReferenceId == 0 {
 		return nil, errors.New("bad type")
@@ -73,7 +73,7 @@ func (s *session) encodeStream(v reflect.Value) ([]byte, error) {
 	return buf[:written], nil
 }
 
-func (s *session) decodeStream(v reflect.Value, b []byte) error {
+func (s *SpdyTransport) decodeStream(v reflect.Value, b []byte) error {
 	referenceId, readN := binary.Uvarint(b)
 	if readN == 0 {
 		return errors.New("bad reference id")
@@ -87,7 +87,7 @@ func (s *session) decodeStream(v reflect.Value, b []byte) error {
 	return nil
 }
 
-func (s *session) waitConn(network, local, remote string, timeout time.Duration) (net.Conn, error) {
+func (s *SpdyTransport) waitConn(network, local, remote string, timeout time.Duration) (net.Conn, error) {
 	timeoutChan := time.After(timeout)
 	connChan := make(chan net.Conn)
 
@@ -173,7 +173,7 @@ func decodeString3(b []byte) (string, string, string, error) {
 	return s1, s2, s3, nil
 }
 
-func (s *session) encodeNetConn(v reflect.Value) ([]byte, error) {
+func (s *SpdyTransport) encodeNetConn(v reflect.Value) ([]byte, error) {
 	var conn net.Conn
 	switch t := v.Interface().(type) {
 	case net.TCPConn:
@@ -190,7 +190,7 @@ func (s *session) encodeNetConn(v reflect.Value) ([]byte, error) {
 	return encodeString3(conn.LocalAddr().Network(), conn.RemoteAddr().String(), conn.LocalAddr().String())
 }
 
-func (s *session) decodeNetConn(v reflect.Value, b []byte) error {
+func (s *SpdyTransport) decodeNetConn(v reflect.Value, b []byte) error {
 	network, local, remote, err := decodeString3(b)
 	if err != nil {
 		return err
@@ -205,7 +205,7 @@ func (s *session) decodeNetConn(v reflect.Value, b []byte) error {
 	return nil
 }
 
-func (s *session) initializeHandler() *codec.MsgpackHandle {
+func (s *SpdyTransport) initializeHandler() *codec.MsgpackHandle {
 	mh := &codec.MsgpackHandle{WriteExt: true}
 	mh.RawToString = true
 	err := mh.AddExt(reflect.TypeOf(channel{}), 0x01, s.encodeChannel, s.decodeChannel)

spdy/listener.go

@@ -2,38 +2,36 @@ package spdy
 
 import (
 	"net"
-
-	"github.com/docker/libchan"
 )
 
-// SessionListener is a listener which accepts new
-// connections angi rd spawns spdy sessions.
-type SessionListener struct {
+// TransportListener is a listener which accepts new
+// connections angi rd spawns spdy transports.
+type TransportListener struct {
 	listener net.Listener
 	auth     Authenticator
 }
 
-// NewSessionListener creates a new listen session using
+// NewTransportListener creates a new listen transport using
 // a network listeners and function to authenticate
-// new connections.  SessionListener expects tls session
+// new connections.  TransportListener expects tls session
 // handling to occur by the authenticator or the listener,
-// SessionListener will not perform tls handshakes.
-func NewSessionListener(listener net.Listener, auth Authenticator) (*SessionListener, error) {
-	return &SessionListener{
+// TransportListener will not perform tls handshakes.
+func NewTransportListener(listener net.Listener, auth Authenticator) (*TransportListener, error) {
+	return &TransportListener{
 		listener: listener,
 		auth:     auth,
 	}, nil
 }
 
 // Close closes the underlying listener
-func (l *SessionListener) Close() error {
+func (l *TransportListener) Close() error {
 	return l.listener.Close()
 }
 
-// AcceptSessions waits for a new network connections
+// AcceptTransport waits for a new network connections
 // and creates a new stream.  Connections which fail
 // authentication will not be returned.
-func (l *SessionListener) AcceptSession() (libchan.Transport, error) {
+func (l *TransportListener) AcceptTransport() (*SpdyTransport, error) {
 	for {
 		conn, err := l.listener.Accept()
 		if err != nil {

spdy/pipe.go

@@ -8,12 +8,12 @@ import (
 )
 
 type pipeSender struct {
-	session *session
+	session *SpdyTransport
 	sender  libchan.Sender
 }
 
 type pipeReceiver struct {
-	session  *session
+	session  *SpdyTransport
 	receiver libchan.Receiver
 }
 

spdy/session.go

@@ -24,9 +24,9 @@ var (
 	ErrWrongDirection = errors.New("Wrong channel direction")
 )
 
-// session is a transport session on top of a network
+// SpdyTransport is a transport session on top of a network
 // connection using spdy.
-type session struct {
+type SpdyTransport struct {
 	conn    *spdystream.Connection
 	handler codec.Handle
 
@@ -44,32 +44,32 @@ type session struct {
 
 type channel struct {
 	stream    *spdystream.Stream
-	session   *session
+	session   *SpdyTransport
 	direction direction
 }
 
-// NewClientSession creates a new stream session from the
+// NewClientTransport creates a new stream transport from the
 // provided network connection. The network connection is
 // expected to already provide a tls session.
-func NewClientSession(conn net.Conn) (libchan.Transport, error) {
+func NewClientTransport(conn net.Conn) (*SpdyTransport, error) {
 	return newSession(conn, false)
 }
 
-// NewServerSession creates a new stream session from the
+// NewServerTransport creates a new stream transport from the
 // provided network connection. The network connection is
 // expected to already have completed the tls handshake.
-func NewServerSession(conn net.Conn) (libchan.Transport, error) {
+func NewServerTransport(conn net.Conn) (*SpdyTransport, error) {
 	return newSession(conn, true)
 }
 
-func newSession(conn net.Conn, server bool) (*session, error) {
+func newSession(conn net.Conn, server bool) (*SpdyTransport, error) {
 	var referenceCounter uint64
 	if server {
 		referenceCounter = 2
 	} else {
 		referenceCounter = 1
 	}
-	session := &session{
+	session := &SpdyTransport{
 		streamChan:       make(chan *spdystream.Stream),
 		referenceCounter: referenceCounter,
 		byteStreamC:      sync.NewCond(new(sync.Mutex)),
@@ -91,7 +91,7 @@ func newSession(conn net.Conn, server bool) (*session, error) {
 	return session, nil
 }
 
-func (s *session) newStreamHandler(stream *spdystream.Stream) {
+func (s *SpdyTransport) newStreamHandler(stream *spdystream.Stream) {
 	referenceIdString := stream.Headers().Get("libchan-ref")
 
 	returnHeaders := http.Header{}
@@ -120,7 +120,7 @@ func (s *session) newStreamHandler(stream *spdystream.Stream) {
 	stream.SendReply(returnHeaders, finish)
 }
 
-func (s *session) getByteStream(referenceId uint64) *byteStream {
+func (s *SpdyTransport) getByteStream(referenceId uint64) *byteStream {
 	s.byteStreamC.L.Lock()
 	bs, ok := s.byteStreams[referenceId]
 	if !ok {
@@ -131,7 +131,7 @@ func (s *session) getByteStream(referenceId uint64) *byteStream {
 	return bs
 }
 
-func (s *session) dial(referenceId uint64) (*byteStream, error) {
+func (s *SpdyTransport) dial(referenceId uint64) (*byteStream, error) {
 	headers := http.Header{}
 	headers.Set("libchan-ref", strconv.FormatUint(referenceId, 10))
 	stream, streamErr := s.conn.CreateStream(headers, nil, false)
@@ -145,7 +145,7 @@ func (s *session) dial(referenceId uint64) (*byteStream, error) {
 	return bs, nil
 }
 
-func (s *session) createByteStream() (io.ReadWriteCloser, error) {
+func (s *SpdyTransport) createByteStream() (io.ReadWriteCloser, error) {
 	s.referenceLock.Lock()
 	referenceId := s.referenceCounter
 	s.referenceCounter = referenceId + 2
@@ -172,7 +172,14 @@ func addrKey(local, remote string) string {
 	return string(b)
 }
 
-func (s *session) RegisterConn(conn net.Conn) error {
+// RegisterConn registers a network connection to be used
+// by inbound messages referring to the connection
+// with the registered connection's local and remote address.
+// Note: a connection does not need to be registered before
+// being sent in a message, but does need to be registered
+// to by the receiver of a message. If registration should be
+// automatic, register a listener instead.
+func (s *SpdyTransport) RegisterConn(conn net.Conn) error {
 	s.netConnC.L.Lock()
 	defer s.netConnC.L.Unlock()
 	networkType, ok := s.networks[conn.LocalAddr().Network()]
@@ -186,7 +193,9 @@ func (s *session) RegisterConn(conn net.Conn) error {
 	return nil
 }
 
-func (s *session) RegisterListener(listener net.Listener) {
+// RegisterListener accepts all connections from the listener
+// and immediately registers them.
+func (s *SpdyTransport) RegisterListener(listener net.Listener) {
 	go func() {
 		for {
 			conn, err := listener.Accept()
@@ -203,18 +212,24 @@ func (s *session) RegisterListener(listener net.Listener) {
 	}()
 }
 
-func (s *session) Unregister(net.Conn) {
+// Unregister removes the connection from the list of known
+// connections. This should be called when a connection is
+// closed and no longer expected in inbound messages.
+// Failure to unregister connections will increase memory
+// usage since the transport is not notified of closed
+// connections to automatically unregister.
+func (s *SpdyTransport) Unregister(net.Conn) {
 
 }
 
-func (s *session) Close() error {
+func (s *SpdyTransport) Close() error {
 	return s.conn.Close()
 }
 
 // NewSendChannel creates and returns a new send channel.  The receive
 // end will get picked up on the remote end through the remote calling
 // WaitReceiveChannel.
-func (s *session) NewSendChannel() (libchan.Sender, error) {
+func (s *SpdyTransport) NewSendChannel() (libchan.Sender, error) {
 	stream, streamErr := s.conn.CreateStream(http.Header{}, nil, false)
 	if streamErr != nil {
 		return nil, streamErr
@@ -224,7 +239,7 @@ func (s *session) NewSendChannel() (libchan.Sender, error) {
 
 // WaitReceiveChannel waits for a new channel be created by a remote
 // call to NewSendChannel.
-func (s *session) WaitReceiveChannel() (libchan.Receiver, error) {
+func (s *SpdyTransport) WaitReceiveChannel() (libchan.Receiver, error) {
 	stream, ok := <-s.streamChan
 	if !ok {
 		return nil, io.EOF