Merge pull request #54 from dmcgowan/channel_reference_id

Update channel passing to use reference ids instead of stream ids

Author: dmcgowan

spdy/encode.go

@@ -10,6 +10,28 @@ import (
 	"github.com/dmcgowan/go/codec"
 )
 
+func decodeReferenceID(b []byte) (referenceID uint64, err error) {
+	if len(b) == 8 {
+		referenceID = binary.BigEndian.Uint64(b)
+	} else if len(b) == 4 {
+		referenceID = uint64(binary.BigEndian.Uint32(b))
+	} else {
+		err = errors.New("bad reference id")
+	}
+	return
+}
+
+func encodeReferenceID(b []byte, referenceID uint64) (n int) {
+	if referenceID > 0xffffffff {
+		binary.BigEndian.PutUint64(b, referenceID)
+		n = 8
+	} else {
+		binary.BigEndian.PutUint32(b, uint32(referenceID))
+		n = 4
+	}
+	return
+}
+
 func (s *Transport) encodeChannel(v reflect.Value) ([]byte, error) {
 	rc := v.Interface().(channel)
 	if rc.stream == nil {
@@ -20,19 +42,15 @@ func (s *Transport) encodeChannel(v reflect.Value) ([]byte, error) {
 	}
 
 	// Get stream identifier?
-	streamID := rc.stream.Identifier()
-	var buf [9]byte
+	buf := make([]byte, 9)
 	if rc.direction == inbound {
 		buf[0] = 0x02 // Reverse direction
 	} else if rc.direction == outbound {
 		buf[0] = 0x01 // Reverse direction
 	} else {
 		return nil, errors.New("invalid direction")
 	}
-	written := binary.PutUvarint(buf[1:], uint64(streamID))
-	if written > 4 {
-		return nil, errors.New("wrote unexpected stream id size")
-	}
+	written := encodeReferenceID(buf[1:], rc.referenceID)
 	return buf[:(written + 1)], nil
 }
 
@@ -46,18 +64,16 @@ func (s *Transport) decodeChannel(v reflect.Value, b []byte) error {
 	} else {
 		return errors.New("unexpected direction")
 	}
-
-	streamID, readN := binary.Uvarint(b[1:])
-	if readN > 4 {
-		return errors.New("read unexpected stream id size")
+	referenceID, err := decodeReferenceID(b[1:])
+	if err != nil {
+		return err
 	}
-	stream := s.conn.FindStream(uint32(streamID))
-	if stream == nil {
-		return errors.New("stream does not exist")
+
+	c := s.getChannel(referenceID)
+	if c == nil {
+		return errors.New("channel does not exist")
 	}
-	rc.session = s
-	rc.stream = stream
-	v.Set(reflect.ValueOf(rc))
+	v.Set(reflect.ValueOf(*c))
 
 	return nil
 }
@@ -84,15 +100,15 @@ func (s *Transport) encodeStream(v reflect.Value) ([]byte, error) {
 		return nil, errors.New("bad type")
 	}
 	var buf [8]byte
-	written := binary.PutUvarint(buf[:], uint64(bs.referenceID))
+	written := encodeReferenceID(buf[:], bs.referenceID)
 
 	return buf[:written], nil
 }
 
 func (s *Transport) decodeStream(v reflect.Value, b []byte) error {
-	referenceID, readN := binary.Uvarint(b)
-	if readN == 0 {
-		return errors.New("bad reference id")
+	referenceID, err := decodeReferenceID(b)
+	if err != nil {
+		return err
 	}
 
 	bs := s.getByteStream(referenceID)

spdy/session.go

@@ -328,9 +328,11 @@ func (c *channel) createSubChannel(direction direction) (libchan.Sender, libchan
 		return nil, nil, streamErr
 	}
 	subChannel := &channel{
-		stream:    stream,
-		session:   c.session,
-		direction: direction,
+		referenceID: referenceID,
+		parentID:    c.referenceID,
+		stream:      stream,
+		session:     c.session,
+		direction:   direction,
 	}
 
 	c.session.channelC.L.Lock()

spdy/session_test.go

@@ -284,6 +284,156 @@ func TestWrappedByteStreams(t *testing.T) {
 	SpawnClientServerTest(t, "localhost:12943", ClientSendWrapper(client), ServerReceiveWrapper(server))
 }
 
+type ReceiverMessage struct {
+	Message  string
+	Receiver libchan.Receiver
+}
+
+func TestSubChannel(t *testing.T) {
+	client := func(t *testing.T, sender libchan.Sender, s *Transport) {
+		remote1, send1 := libchan.Pipe()
+		m1 := &ReceiverMessage{
+			Message:  "WithReceiver",
+			Receiver: remote1,
+		}
+		if sendErr := sender.Send(m1); sendErr != nil {
+			t.Fatalf("Error sending ReceiverMessage: %s", sendErr)
+		}
+
+		remote2, send2 := libchan.Pipe()
+		m2 := &ReceiverMessage{
+			Message:  "Nested",
+			Receiver: remote2,
+		}
+		if sendErr := send1.Send(m2); sendErr != nil {
+			t.Fatalf("Error sending ReceiverMessage: %s", sendErr)
+		}
+
+		m3 := &SimpleMessage{"This is a simple message"}
+		if sendErr := send2.Send(m3); sendErr != nil {
+			t.Fatalf("Error sending simple message: %s", sendErr)
+		}
+
+		if closeErr := send1.Close(); closeErr != nil {
+			t.Fatalf("Error closing send1: %s", closeErr)
+		}
+
+		if closeErr := send2.Close(); closeErr != nil {
+			t.Fatalf("Error closing send2: %s", closeErr)
+		}
+	}
+	server := func(t *testing.T, receiver libchan.Receiver, s *Transport) {
+		m1 := &ReceiverMessage{}
+		if receiveErr := receiver.Receive(m1); receiveErr != nil {
+			t.Fatalf("Error receiving ReceiverMessage: %s", receiveErr)
+		}
+
+		if expected := "WithReceiver"; m1.Message != expected {
+			t.Fatalf("Unexpected message\n\tExpected: %s\n\tActual: %s", expected, m1.Message)
+		}
+
+		if m1.Receiver == nil {
+			t.Fatalf("Receiver is nil")
+		}
+
+		m2 := &ReceiverMessage{}
+		if receiveErr := m1.Receiver.Receive(m2); receiveErr != nil {
+			t.Fatalf("Error receiving ReceiverMessage: %s", receiveErr)
+		}
+		if expected := "Nested"; m2.Message != expected {
+			t.Fatalf("Unexpected message value:\n\tExpected: %s\n\tActual: %s", expected, m2.Message)
+		}
+
+		if m2.Receiver == nil {
+			t.Fatalf("Receiver is nil")
+		}
+
+		m3 := &SimpleMessage{}
+		if receiverErr := m2.Receiver.Receive(m3); receiverErr != nil {
+			t.Fatalf("Error receiving SimpleMessage: %s", receiverErr)
+		}
+		if expected := "This is a simple message"; m3.Message != expected {
+			t.Fatalf("Unexpected message value:\n\tExpected: %s\n\tActual: %s", expected, m3.Message)
+		}
+	}
+	SpawnClientServerTest(t, "localhost:12845", ClientSendWrapper(client), ServerReceiveWrapper(server))
+}
+
+type SenderMessage struct {
+	Message string
+	Sender  libchan.Sender
+}
+
+func TestSenderSubChannel(t *testing.T) {
+	client := func(t *testing.T, sender libchan.Sender, s *Transport) {
+		recv, remote1 := libchan.Pipe()
+		m1 := &SenderMessage{
+			Message: "WithSender",
+			Sender:  remote1,
+		}
+		if sendErr := sender.Send(m1); sendErr != nil {
+			t.Fatalf("Error sending SenderMessage: %s", sendErr)
+		}
+
+		m2 := &SenderMessage{}
+		if receiveErr := recv.Receive(m2); receiveErr != nil {
+			t.Fatalf("Error receiving ReceiverMessage: %s", receiveErr)
+		}
+		if expected := "Nested"; m2.Message != expected {
+			t.Fatalf("Unexpected message value:\n\tExpected: %s\n\tActual: %s", expected, m2.Message)
+		}
+
+		if m2.Sender == nil {
+			t.Fatalf("Receiver is nil")
+		}
+
+		m3 := &SimpleMessage{"This is a simple message"}
+		if sendErr := m2.Sender.Send(m3); sendErr != nil {
+			t.Fatalf("Error sending simple message: %s", sendErr)
+		}
+
+		if closeErr := m2.Sender.Close(); closeErr != nil {
+			t.Fatalf("Error closing send2: %s", closeErr)
+		}
+	}
+	server := func(t *testing.T, receiver libchan.Receiver, s *Transport) {
+		m1 := &SenderMessage{}
+		if receiveErr := receiver.Receive(m1); receiveErr != nil {
+			t.Fatalf("Error receiving SenderMessage: %s", receiveErr)
+		}
+
+		if expected := "WithSender"; m1.Message != expected {
+			t.Fatalf("Unexpected message\n\tExpected: %s\n\tActual: %s", expected, m1.Message)
+		}
+
+		if m1.Sender == nil {
+			t.Fatalf("Receiver is nil")
+		}
+
+		recv, remote := libchan.Pipe()
+		m2 := &SenderMessage{
+			Message: "Nested",
+			Sender:  remote,
+		}
+		if sendErr := m1.Sender.Send(m2); sendErr != nil {
+			t.Fatalf("Error sending SenderMessage: %s", sendErr)
+		}
+
+		m3 := &SimpleMessage{}
+		if receiverErr := recv.Receive(m3); receiverErr != nil {
+			t.Fatalf("Error receiving SimpleMessage: %s", receiverErr)
+		}
+		if expected := "This is a simple message"; m3.Message != expected {
+			t.Fatalf("Unexpected message value:\n\tExpected: %s\n\tActual: %s", expected, m3.Message)
+		}
+
+		if closeErr := m1.Sender.Close(); closeErr != nil {
+			t.Fatalf("Error closing send1: %s", closeErr)
+		}
+
+	}
+	SpawnClientServerTest(t, "localhost:12845", ClientSendWrapper(client), ServerReceiveWrapper(server))
+}
 func ClientSendWrapper(f func(t *testing.T, c libchan.Sender, s *Transport)) ClientRoutine {
 	return func(t *testing.T, server string) {
 		conn, connErr := net.Dial("tcp", server)