Update Subscribe Request format to PR#277

see moq-wg/moq-transport#277

examples/chat/client.go

@@ -55,7 +55,7 @@ func NewClient(p *moqtransport.Peer) (*Client, error) {
 	c.peer.OnAnnouncement(func(s string) error {
 		return nil
 	})
-	c.peer.OnSubscription(func(trackname string, st *moqtransport.SendTrack) (uint64, time.Duration, error) {
+	c.peer.OnSubscription(func(_, trackname string, st *moqtransport.SendTrack) (uint64, time.Duration, error) {
 		namespace := strings.SplitN(trackname, "/", 3)
 		if len(namespace) < 2 {
 			return 0, 0, errors.New("invalid trackname")
@@ -89,7 +89,8 @@ func (c *Client) handleCatalogDeltas(roomID, username string, catalogTrack *moqt
 			if p == username {
 				continue
 			}
-			t, err := c.peer.Subscribe(fmt.Sprintf("moq-chat/%v/%v", roomID, p))
+			fullname := fmt.Sprintf("moq-chat/%v/%v", roomID, p)
+			t, err := c.peer.Subscribe(fullname, fullname)
 			if err != nil {
 				log.Fatal(err)
 			}
@@ -120,7 +121,7 @@ func (c *Client) joinRoom(roomID, username string) {
 	if err := c.peer.Announce(fmt.Sprintf("moq-chat/%v/%v", roomID, username)); err != nil {
 		log.Fatal(err)
 	}
-	catalogTrack, err := c.peer.Subscribe(fmt.Sprintf("moq-chat/%v", roomID))
+	catalogTrack, err := c.peer.Subscribe("", fmt.Sprintf("moq-chat/%v", roomID))
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -139,7 +140,8 @@ func (c *Client) joinRoom(roomID, username string) {
 		if p == username {
 			continue
 		}
-		t, err := c.peer.Subscribe(fmt.Sprintf("moq-chat/%v/%v", roomID, p))
+		fullname := fmt.Sprintf("moq-chat/%v/%v", roomID, p)
+		t, err := c.peer.Subscribe(fullname, fullname)
 		if err != nil {
 			log.Fatal(err)
 		}

examples/chat/room.go

@@ -47,7 +47,8 @@ func (r *room) join(username string, p *moqtransport.Peer) error {
 	if _, ok := r.publishers[username]; ok {
 		return errors.New("username already taken")
 	}
-	t, err := p.Subscribe(fmt.Sprintf("moq-chat/%v/%v", r.id, username))
+	fullname := fmt.Sprintf("moq-chat/%v/%v", r.id, username)
+	t, err := p.Subscribe(fullname, fullname)
 	if err != nil {
 		return err
 	}

examples/chat/server.go

@@ -68,7 +68,7 @@ func (s *Server) peerHandler() moqtransport.PeerHandlerFunc {
 			return s.chatRooms[id].join(name, p)
 		})
 
-		p.OnSubscription(func(trackname string, t *moqtransport.SendTrack) (uint64, time.Duration, error) {
+		p.OnSubscription(func(_, trackname string, t *moqtransport.SendTrack) (uint64, time.Duration, error) {
 			if len(name) == 0 {
 				// Subscribe requires a username which has to be announced
 				// before subscribing

examples/date-client/main.go

@@ -33,8 +33,8 @@ func run(addr string) error {
 		log.Printf("got announcement: %v", s)
 		return nil
 	})
-	p.OnSubscription(func(s string, _ *moqtransport.SendTrack) (uint64, time.Duration, error) {
-		log.Printf("got subscription attempt: %v", s)
+	p.OnSubscription(func(namespace, name string, _ *moqtransport.SendTrack) (uint64, time.Duration, error) {
+		log.Printf("got subscription attempt: %v/%v", namespace, name)
 		return 0, time.Duration(0), nil
 	})
 	go func() {
@@ -43,7 +43,7 @@ func run(addr string) error {
 		}
 	}()
 	log.Println("subscribing")
-	rt, err := p.Subscribe("clock/second")
+	rt, err := p.Subscribe("clock", "second")
 	if err != nil {
 		panic(err)
 	}

examples/date-server/main.go

@@ -49,8 +49,8 @@ func run(addr string, wt bool, certFile, keyFile string) error {
 
 func handler() moqtransport.PeerHandlerFunc {
 	return func(p *moqtransport.Peer) {
-		p.OnSubscription(func(trackname string, t *moqtransport.SendTrack) (uint64, time.Duration, error) {
-			if trackname != "clock/second" {
+		p.OnSubscription(func(namespace, trackname string, t *moqtransport.SendTrack) (uint64, time.Duration, error) {
+			if fmt.Sprintf("%v/%v", namespace, trackname) != "clock/second" {
 				return 0, 0, errors.New("unknown track ID")
 			}
 			go func() {

message.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"log"
 	"time"
 
 	"github.com/quic-go/quic-go/quicvarint"
@@ -97,51 +98,47 @@ type messageReader interface {
 	io.ByteReader
 }
 
-func readNext(reader messageReader) (message, error) {
+func readNext(reader messageReader) (msg message, err error) {
 	mt, err := quicvarint.Read(reader)
 	if err != nil {
 		return nil, err
 	}
 	switch messageType(mt) {
 	case objectMessageLenType:
-		msg, err := parseObjectMessage(reader, mt)
-		return msg, err
+		msg, err = parseObjectMessage(reader, mt)
 	case objectMessageNoLenType:
-		msg, err := parseObjectMessage(reader, mt)
-		return msg, err
+		msg, err = parseObjectMessage(reader, mt)
 	case subscribeRequestMessageType:
-		srm, err := parseSubscribeMessage(reader)
-		return srm, err
+		msg, err = parseSubscribeRequestMessage(reader)
 	case subscribeOkMessageType:
-		som, err := parseSubscribeOkMessage(reader)
-		return som, err
+		msg, err = parseSubscribeOkMessage(reader)
 	case subscribeErrorMessageType:
-		sem, err := parseSubscribeErrorMessage(reader)
-		return sem, err
+		msg, err = parseSubscribeErrorMessage(reader)
 	case announceMessageType:
-		am, err := parseAnnounceMessage(reader)
-		return am, err
+		msg, err = parseAnnounceMessage(reader)
 	case announceOkMessageType:
-		aom, err := parseAnnounceOkMessage(reader)
-		return aom, err
+		msg, err = parseAnnounceOkMessage(reader)
 	case announceErrorMessageType:
-		return parseAnnounceErrorMessage(reader)
+		msg, err = parseAnnounceErrorMessage(reader)
 	case unannounceMessageType:
-		return parseUnannounceMessage(reader)
+		msg, err = parseUnannounceMessage(reader)
 	case unsubscribeMessageType:
-		return parseUnsubscribeMessage(reader)
+		msg, err = parseUnsubscribeMessage(reader)
 	case subscribeFinMessageType:
-		return parseSubscribeFinMessage(reader)
+		msg, err = parseSubscribeFinMessage(reader)
 	case subscribeRstMessageType:
-		return parseSubscribeRstMessage(reader)
+		msg, err = parseSubscribeRstMessage(reader)
 	case goAwayMessageType:
-		return parseGoAwayMessage(reader)
+		msg, err = parseGoAwayMessage(reader)
 	case clientSetupMessageType:
-		return parseClientSetupMessage(reader)
+		msg, err = parseClientSetupMessage(reader)
 	case serverSetupMessageType:
-		return parseServerSetupMessage(reader)
+		msg, err = parseServerSetupMessage(reader)
+	default:
+		return nil, errors.New("unknown message type")
 	}
-	return nil, errors.New("unknown message type")
+	log.Printf("parsed message: %v, err: %v", msg, err)
+	return
 }
 
 type objectMessage struct {
@@ -249,7 +246,10 @@ type clientSetupMessage struct {
 }
 
 func (m clientSetupMessage) String() string {
-	return clientSetupMessageType.String()
+	out := clientSetupMessageType.String()
+	out += fmt.Sprintf("\tSupportedVersions: %v\n", m.supportedVersions)
+	out += fmt.Sprintf("\tSetupParameters: %v\n", m.setupParameters)
+	return out
 }
 
 func (m *clientSetupMessage) append(buf []byte) []byte {
@@ -289,7 +289,10 @@ type serverSetupMessage struct {
 }
 
 func (m serverSetupMessage) String() string {
-	return serverSetupMessageType.String()
+	out := serverSetupMessageType.String()
+	out += fmt.Sprintf("\tSelectedVersions: %v\n", m.selectedVersion)
+	out += fmt.Sprintf("\tSetupParameters: %v\n", m.setupParameters)
+	return out
 }
 
 func (m *serverSetupMessage) append(buf []byte) []byte {
@@ -358,17 +361,19 @@ func parseLocation(r messageReader) (location, error) {
 }
 
 type subscribeRequestMessage struct {
-	fullTrackName string
-	startGroup    location
-	startObject   location
-	endGroup      location
-	endObject     location
-	parameters    parameters
+	trackNamespace string
+	trackName      string
+	startGroup     location
+	startObject    location
+	endGroup       location
+	endObject      location
+	parameters     parameters
 }
 
 func (m subscribeRequestMessage) String() string {
 	out := subscribeRequestMessageType.String()
-	out += fmt.Sprintf("\tFullTrackName: %v\n", m.fullTrackName)
+	out += fmt.Sprintf("\tTrackNamespace: %v\n", m.trackNamespace)
+	out += fmt.Sprintf("\tTrackName: %v\n", m.trackName)
 	out += fmt.Sprintf("\tStartGroup: %v\n", m.startGroup)
 	out += fmt.Sprintf("\tStartObject: %v\n", m.startObject)
 	out += fmt.Sprintf("\tEndGroup: %v\n", m.endGroup)
@@ -380,13 +385,14 @@ func (m subscribeRequestMessage) String() string {
 func (m subscribeRequestMessage) key() messageKey {
 	return messageKey{
 		mt: subscribeRequestMessageType,
-		id: m.fullTrackName,
+		id: fmt.Sprintf("%v/%v", m.trackNamespace, m.trackName),
 	}
 }
 
 func (m *subscribeRequestMessage) append(buf []byte) []byte {
 	buf = quicvarint.Append(buf, uint64(subscribeRequestMessageType))
-	buf = appendVarIntString(buf, m.fullTrackName)
+	buf = appendVarIntString(buf, m.trackNamespace)
+	buf = appendVarIntString(buf, m.trackName)
 	buf = m.startGroup.append(buf)
 	buf = m.startObject.append(buf)
 	buf = m.endGroup.append(buf)
@@ -398,10 +404,14 @@ func (m *subscribeRequestMessage) append(buf []byte) []byte {
 	return buf
 }
 
-func parseSubscribeMessage(r messageReader) (*subscribeRequestMessage, error) {
+func parseSubscribeRequestMessage(r messageReader) (*subscribeRequestMessage, error) {
 	if r == nil {
 		return nil, errInvalidMessageReader
 	}
+	trackNamespace, err := parseVarIntString(r)
+	if err != nil {
+		return nil, err
+	}
 	fullTrackName, err := parseVarIntString(r)
 	if err != nil {
 		return nil, err
@@ -427,12 +437,13 @@ func parseSubscribeMessage(r messageReader) (*subscribeRequestMessage, error) {
 		return nil, err
 	}
 	return &subscribeRequestMessage{
-		fullTrackName: fullTrackName,
-		startGroup:    startGroup,
-		startObject:   startObject,
-		endGroup:      endGroup,
-		endObject:     endObject,
-		parameters:    ps,
+		trackNamespace: trackNamespace,
+		trackName:      fullTrackName,
+		startGroup:     startGroup,
+		startObject:    startObject,
+		endGroup:       endGroup,
+		endObject:      endObject,
+		parameters:     ps,
 	}, nil
 }
 
@@ -455,7 +466,7 @@ func (m subscribeOkMessage) String() string {
 func (m subscribeOkMessage) key() messageKey {
 	return messageKey{
 		mt: subscribeRequestMessageType,
-		id: m.trackName,
+		id: fmt.Sprintf("%v/%v", m.trackNamespace, m.trackName),
 	}
 }
 
@@ -515,7 +526,7 @@ func (m subscribeErrorMessage) String() string {
 func (m subscribeErrorMessage) key() messageKey {
 	return messageKey{
 		mt: subscribeRequestMessageType,
-		id: m.trackName,
+		id: fmt.Sprintf("%v/%v", m.trackNamespace, m.trackName),
 	}
 }