moqtransport

README

Media over QUIC Transport

moqtransport is an implementation of Media over QUIC Transport on top of quic-go and optionally webtransport-go.

Example: Chat Server and Client

The examples directory contains an implementation of MoQ Chat.

To run a simple chat server using MoQ Tranpsort on top of QUIC, run:

go run examples/chat-server/main.go

Then, open a new shell and start a client:

go run examples/chat-client/main.go

The client is a simple interactive shell that reads and writes messages from stdin and stdout (input and output is currently not well synchronized).

Open another shell and run a second client to chat with the first one using the commands join <roomID> <username> to join a room and msg <roomID> <message> to send messages to a room.

To use WebTransport, you need to create a TLS certificate. This can be done using mkcert:

mkcert localhost
mkcert -install

mkcert will generate a localhost.pem and a localhost-key.pem file. If you change the name of the files or use a different host name, you can use the -cert and -key flags of the server command to point it to the correct files.

Now start the server and client with the -webtransport and -addr flags to run MoQ Transport on top of WebTransport:

Server:

go run examples/chat-server/main.go -webtransport

Client:

go run examples/chat-client/main.go -webtransport -addr https://localhost:8080/moq

Documentation

Index

• Constants
• func SetLogHandler(handler slog.Handler)
• type Announcement
  • func (a *Announcement) Namespace() string
• type AnnouncementHandler
• type AnnouncementHandlerFunc
  • func (f AnnouncementHandlerFunc) HandleAnnouncement(s *Session, a *Announcement, arw AnnouncementResponseWriter)
• type AnnouncementResponseWriter
• type ApplicationError
  • func (e ApplicationError) Error() string
• type Connection
• type LocalTrack
  • func NewLocalTrack(id uint64, namespace, trackname string) *LocalTrack
  • func (t *LocalTrack) Close() error
  • func (t *LocalTrack) SubscriberCount() int
  • func (t *LocalTrack) WriteObject(ctx context.Context, o Object) error
• type Location
• type LocationMode
• type Object
• type ObjectForwardingPreference
• type ObjectWriter
• type ProtocolError
  • func (e ProtocolError) Code() uint64
  • func (e ProtocolError) Error() string
• type ReceiveStream
• type RemoteTrack
  • func (s *RemoteTrack) ReadObject(ctx context.Context) (Object, error)
  • func (s *RemoteTrack) Unsubscribe()
• type Role
• type SendStream
• type Session
  • func (s *Session) AddLocalTrack(t *LocalTrack) error
  • func (s *Session) Announce(ctx context.Context, namespace string) error
  • func (s *Session) Close() error
  • func (s *Session) CloseWithError(code uint64, msg string) error
  • func (s *Session) RunClient() error
  • func (s *Session) RunServer(ctx context.Context) error
  • func (s *Session) Subscribe(ctx context.Context, subscribeID, trackAlias uint64, ...) (*RemoteTrack, error)
• type Stream
• type Subscription
• type SubscriptionHandler
• type SubscriptionHandlerFunc
  • func (f SubscriptionHandlerFunc) HandleSubscription(se *Session, su *Subscription, srw SubscriptionResponseWriter)
• type SubscriptionResponseWriter

Constants

const (
	ErrorCodeNoError                 = 0x00
	ErrorCodeInternal                = 0x01
	ErrorCodeUnauthorized            = 0x02
	ErrorCodeProtocolViolation       = 0x03
	ErrorCodeDuplicateTrackAlias     = 0x04
	ErrorCodeParameterLengthMismatch = 0x05
	ErrorCodeGoAwayTimeout           = 0x10

	// Errors not included in current draft
	ErrorCodeUnsupportedVersion = 0xff01
	ErrorCodeTrackNotFound      = 0xff02
)

const (
	SubscribeErrorInternal        = 0x00
	SubscribeErrorInvalidRange    = 0x01
	SubscribeErrorRetryTrackAlias = 0x02

	// TODO: These are not specified yet, but seem useful
	SubscribeErrorUnknownTrack = 0x03
)

const (
	SubscribeStatusUnsubscribed      = 0x00
	SubscribeStatusInternalError     = 0x01
	SubscribeStatusUnauthorized      = 0x02
	SubscribeStatusTrackEnded        = 0x03
	SubscribeStatusSubscriptionEnded = 0x04
	SubscribeStatusGoingAway         = 0x05
	SubscribeStatusExpired           = 0x06
)

const (
	DRAFT_IETF_MOQ_TRANSPORT_00 = 0xff000000
	DRAFT_IETF_MOQ_TRANSPORT_01 = 0xff000001
	DRAFT_IETF_MOQ_TRANSPORT_02 = 0xff000002
	DRAFT_IETF_MOQ_TRANSPORT_03 = 0xff000003

	CURRENT_VERSION = DRAFT_IETF_MOQ_TRANSPORT_03
)

Functions

func SetLogHandler

func SetLogHandler(handler slog.Handler)

Types

type Announcement

type Announcement struct {
	// contains filtered or unexported fields
}

func (*Announcement) Namespace

func (a *Announcement) Namespace() string

type AnnouncementHandler

type AnnouncementHandler interface {
	HandleAnnouncement(*Session, *Announcement, AnnouncementResponseWriter)
}

type AnnouncementHandlerFunc

type AnnouncementHandlerFunc func(*Session, *Announcement, AnnouncementResponseWriter)

func (AnnouncementHandlerFunc) HandleAnnouncement

func (f AnnouncementHandlerFunc) HandleAnnouncement(s *Session, a *Announcement, arw AnnouncementResponseWriter)

type AnnouncementResponseWriter

type AnnouncementResponseWriter interface {
	Accept()
	Reject(code uint64, reason string)
}

type ApplicationError

type ApplicationError struct {
	// contains filtered or unexported fields
}

func (ApplicationError) Error

func (e ApplicationError) Error() string

type Connection

type Connection interface {
	OpenStream() (Stream, error)
	OpenStreamSync(context.Context) (Stream, error)
	OpenUniStream() (SendStream, error)
	OpenUniStreamSync(context.Context) (SendStream, error)
	AcceptStream(context.Context) (Stream, error)
	AcceptUniStream(context.Context) (ReceiveStream, error)
	SendDatagram([]byte) error
	ReceiveDatagram(context.Context) ([]byte, error)
	CloseWithError(uint64, string) error
}

type LocalTrack

type LocalTrack struct {
	ID        uint64
	Namespace string
	Name      string
	// contains filtered or unexported fields
}

A LocalTrack is a local media source. Writing objects to the track will relay the objects to all subscribers. All methods are safe for concurrent use. Ordering of objects is only guaranteed within MultiObjectStreams. LocalTracks must be created with NewLocalTrack to ensure proper initialization.

func NewLocalTrack

func NewLocalTrack(id uint64, namespace, trackname string) *LocalTrack

NewLocalTrack creates a new LocalTrack

func (*LocalTrack) Close

func (t *LocalTrack) Close() error

func (*LocalTrack) SubscriberCount

func (t *LocalTrack) SubscriberCount() int

func (*LocalTrack) WriteObject

func (t *LocalTrack) WriteObject(ctx context.Context, o Object) error

WriteObject adds an object to the track

type Location

type Location struct {
	Mode  LocationMode
	Value uint64
}

type LocationMode

type LocationMode int

const (
	LocationModeNone LocationMode = iota
	LocationModeAbsolute
	LocationModeRelativePrevious
	LocationModeRelativeNext
)

type Object

type Object struct {
	GroupID              uint64
	ObjectID             uint64
	ObjectSendOrder      uint64
	ForwardingPreference ObjectForwardingPreference

	Payload []byte
}

type ObjectForwardingPreference

type ObjectForwardingPreference int

const (
	ObjectForwardingPreferenceDatagram ObjectForwardingPreference = iota
	ObjectForwardingPreferenceStream
	ObjectForwardingPreferenceStreamGroup
	ObjectForwardingPreferenceStreamTrack
)

type ObjectWriter

type ObjectWriter interface {
	WriteObject(Object) error
	io.Closer
}

An ObjectWriter allows sending objects using.

type ProtocolError

type ProtocolError struct {
	// contains filtered or unexported fields
}

func (ProtocolError) Code

func (e ProtocolError) Code() uint64

func (ProtocolError) Error

func (e ProtocolError) Error() string

type ReceiveStream

type ReceiveStream interface {
	io.Reader
}

type RemoteTrack

type RemoteTrack struct {
	// contains filtered or unexported fields
}

func (*RemoteTrack) ReadObject

func (s *RemoteTrack) ReadObject(ctx context.Context) (Object, error)

func (*RemoteTrack) Unsubscribe

func (s *RemoteTrack) Unsubscribe()

type Role

type Role uint64

const (
	RolePublisher Role = iota + 1
	RoleSubscriber
	RolePubSub
)

type SendStream

type SendStream interface {
	io.WriteCloser
}

type Session

type Session struct {
	Conn                Connection
	EnableDatagrams     bool
	LocalRole           Role
	RemoteRole          Role
	AnnouncementHandler AnnouncementHandler
	SubscriptionHandler SubscriptionHandler
	// contains filtered or unexported fields
}

func (*Session) AddLocalTrack

func (s *Session) AddLocalTrack(t *LocalTrack) error

func (*Session) Announce

func (s *Session) Announce(ctx context.Context, namespace string) error

func (*Session) Close

func (s *Session) Close() error

func (*Session) CloseWithError

func (s *Session) CloseWithError(code uint64, msg string) error

func (*Session) RunClient

func (s *Session) RunClient() error

func (*Session) RunServer

func (s *Session) RunServer(ctx context.Context) error

func (*Session) Subscribe

func (s *Session) Subscribe(ctx context.Context, subscribeID, trackAlias uint64, namespace, trackname, auth string) (*RemoteTrack, error)

type Stream

type Stream interface {
	ReceiveStream
	SendStream
}

type Subscription

type Subscription struct {
	ID            uint64
	TrackAlias    uint64
	Namespace     string
	TrackName     string
	Authorization string
}

type SubscriptionHandler

type SubscriptionHandler interface {
	HandleSubscription(*Session, *Subscription, SubscriptionResponseWriter)
}

type SubscriptionHandlerFunc

type SubscriptionHandlerFunc func(*Session, *Subscription, SubscriptionResponseWriter)

func (SubscriptionHandlerFunc) HandleSubscription

func (f SubscriptionHandlerFunc) HandleSubscription(se *Session, su *Subscription, srw SubscriptionResponseWriter)

type SubscriptionResponseWriter

type SubscriptionResponseWriter interface {
	Accept(*LocalTrack)
	Reject(code uint64, reason string)
}