moqt

package
v0.0.0-...-f4df4ba Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Apr 22, 2025 License: MIT Imports: 23 Imported by: 0

README

gomoqt

Overview

This is an implementation of Media over QUIC Transfork.

Specification

https://kixelated.github.io/moq-drafts/draft-lcurley-moq-transfork.html

Implementation Status

Section Implemented Tested
2. Data Model
2.1. Frame ✅ ✅
2.3. Group ✅ ✅
2.4. Track ✅ ✅
2.4.1. Track Naming and Scopes 🚧 ❌
2.4.2. Scope 🚧 ❌
2.4.3. Connection URL 🚧 ❌
3. Sessions
3.1. Session establishment ✅ ✅
3.1.1. WebTransport ✅ ✅
3.1.2. QUIC ✅ ❌
3.2. Version and Extension Negotiation ✅ ❌
3.3. Session initialization ✅ ✅
3.4. Stream Cancellation 🚧 ❌
3.5. Termination ✅ ✅
3.6. Migration 🚧 ❌
4. Data Transmittions
4.1 Track Priority Control 🚧 ❌
4.2 Group Order Control 🚧 ❌
4.3 Cache 🚧 ❌
5. Relays
5.1. Subscriber Interactions ✅ ✅
5.1.1. Graceful Publisher Relay Switchover ❌ ❌
5.2. Publisher Interactions ✅ ✅
5.2.1. Graceful Publisher Network Switchover ❌ ❌
5.2.2. Graceful Publisher Relay Switchover ❌ ❌
5.3. Relay Object Handling 🚧 ❌
Control Streams
6.1. Session Stream ✅ ✅
6.2. Announce Stream ✅ ✅
6.3. Subscribe Stream ✅ ✅
6.4. Fetch Stream ✅ ✅
6.5. Info Stream ✅ ✅
Control Messages
6.1. Parameters 🚧 ✅
6.1.1. Version Specific Parameters ✅ ✅
6.2. SESSION_CLIENT ✅ ✅
6.2. SESSION_SERVER ✅ ✅
6.2. SESSION_UPDATE ✅ ✅
6.2.1. Versions ✅ ✅
6.2.2. Setup Parameters ✅ ✅
6.3. ANNOUNCE_PLEASE ✅ ✅
6.4. ANNOUNCE ✅ ✅
6.5. SUBSCRIBE ✅ ✅
6.6. SUBSCRIBE_UPDATE ✅ ✅
6.7. SUBSCRIBE_GAP ❌ ❌
6.8. INFO ✅ ✅
6.9. INFO_PLEASE ✅ ✅
Data Stream
7.3. Group Streams ✅ ✅
Data Message
7.2. GROUP ✅ ✅
7.2. FRAME ✅ ✅
Security Considerations
8.1. Resource Exhaustion ❌ ❌

Interoperablity test

We haven't conducted interoperability testing with other implementations yet

TODO

  • Interoperability
  • Scheduling
  • LOC (Low Overhead Container)
  • Common Catalog Format for moq
  • sync.Pool

Documentation

Index

Constants

View Source 
const NextProtoMOQ = "moq-00"

Variables

View Source 
var (
	ErrInternalError = &Error{internal.ErrInternalError}

	ErrUnauthorizedError = &Error{internal.ErrUnauthorizedError} // TODO: Use this error

	ErrTrackDoesNotExist = &Error{internal.ErrTrackDoesNotExist}

	ErrDuplicatedTrack = &Error{internal.ErrDuplicatedTrack}

	ErrInvalidRange = &Error{internal.ErrInvalidRange}

	ErrDuplicatedSubscribeID = &Error{internal.ErrDuplicatedSubscribeID} // TODO: Use this error

	ErrEndedTrack = &Error{internal.ErrEndedTrack}

	ErrClosedTrack = &Error{internal.ErrClosedTrack}

	NoErrTerminate = &Error{internal.NoErrTerminate}

	ErrProtocolViolation = &Error{internal.ErrProtocolViolation}

	ErrGroupRejected = &Error{internal.ErrGroupRejected}

	ErrGroupOutOfRange = &Error{internal.ErrGroupOutOfRange}

	ErrGroupExpired = &Error{internal.ErrGroupExpired}

	ErrClosedGroup = &Error{internal.ErrClosedGroup}
)
View Source 
var DefaultFrameSize = 2048
View Source 
var DefaultGroupBufferSize = defaultBufferSize
View Source 
var ErrParameterNotFound = errors.New("parameter not found")
View Source 
var NoSetupExtensions = func(req Parameters) (rsp Parameters, err error) {
	return Parameters{}, nil
}

Functions

This section is empty.

Types

type AnnounceConfig 

type AnnounceConfig struct {
	TrackPrefix string
}

func (AnnounceConfig) String 

func (ac AnnounceConfig) String() string

type AnnounceError 

type AnnounceError interface {
	error
	AnnounceErrorCode() AnnounceErrorCode
}

type AnnounceErrorCode 

type AnnounceErrorCode uint64

type Announcement 

type Announcement struct {

	/*
	 *
	 */
	TrackPath TrackPath
	// contains filtered or unexported fields
}

func NewAnnouncement 

func NewAnnouncement(trackPath TrackPath) *Announcement

func (Announcement) IsActive 

func (a Announcement) IsActive() bool

func (Announcement) IsEnded 

func (a Announcement) IsEnded() bool

func (Announcement) String 

func (a Announcement) String() string

type AnnouncementReader 

type AnnouncementReader interface {
	NextAnnouncements(context.Context) ([]*Announcement, error)
	AnnounceConfig() AnnounceConfig
	Close() error
	CloseWithError(error) error
}

type AnnouncementWriter 

type AnnouncementWriter interface {
	WriteAnnouncement(announcements []*Announcement) error
	// AnnounceConfig() AnnounceConfig
	Close() error
	CloseWithError(error) error
}

type Client 

type Client struct {
	/*
	 * TLS configuration
	 */
	TLSConfig *tls.Config

	/*
	 * QUIC configuration
	 */
	QUICConfig *quic.Config

	/*
	 * MOQ Configuration
	 */
	Config *Config

	/***/
	SetupExtensions *Parameters

	/*
	 * Logger
	 */
	Logger *slog.Logger
	// contains filtered or unexported fields
}

func (*Client) Dial 

func (c *Client) Dial(urlStr string, ctx context.Context) (Session, SetupResponse, error)

func (*Client) DialQUIC 

func (c *Client) DialQUIC(req SetupRequest, ctx context.Context) (Session, SetupResponse, error)

func (*Client) DialWebTransport 

func (c *Client) DialWebTransport(req SetupRequest, ctx context.Context) (Session, SetupResponse, error)

type Config 

type Config struct {
	// Configurations
	MaxSubscribeID SubscribeID // TODO:

	NewSessionURI string // TODO:

	CheckRoot func(r SetupRequest) bool
}

type Error 

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

func (*Error) As 

func (e *Error) As(target any) bool

func (*Error) Error 

func (e *Error) Error() string

func (*Error) Is 

func (e *Error) Is(target error) bool

func (*Error) Unwrap 

func (e *Error) Unwrap() error

type Frame 

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

func NewFrame 

func NewFrame(b []byte) *Frame

NewFrame creates a new Frame with the specified bytes. The bytes are not copied, so the caller must not modify the bytes after calling this function.

func (Frame) CopyBytes 

func (f Frame) CopyBytes() []byte

Updated CopyBytes method to return a copy of the internal slice.

func (*Frame) Release 

func (f *Frame) Release()

func (Frame) Size 

func (f Frame) Size() int

type FrameSequence 

type FrameSequence uint64

type GroupBuffer 

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

func (*GroupBuffer) GroupSequence 

func (g *GroupBuffer) GroupSequence() GroupSequence

func (*GroupBuffer) Release 

func (g *GroupBuffer) Release()

Release resets the buffer with a new group sequence.

func (*GroupBuffer) SetDeadline 

func (g *GroupBuffer) SetDeadline(t time.Time)

type GroupError 

type GroupError interface {
	error
	GroupErrorCode() GroupErrorCode
}

func NewGroupError 

func NewGroupError(reason string, code GroupErrorCode) GroupError

type GroupErrorCode 

type GroupErrorCode uint64

type GroupOrder 

type GroupOrder byte
const (
	DEFAULT    GroupOrder = 0x0
	ASCENDING  GroupOrder = 0x1
	DESCENDING GroupOrder = 0x2
)

type GroupReader 

type GroupReader interface {
	GroupSequence() GroupSequence
	ReadFrame() (*Frame, error)
	CancelRead(GroupError)
	SetReadDeadline(time.Time) error
}

* Group Reader

type GroupSequence 

type GroupSequence uint64

* Sequence number of a group in a track * When this is integer more than 1, the number means the sequence number. * When this is 0, it indicates the sequence number is currently unknown . * 0 is used to specify "the latest sequence number" or "the final sequence number of an open-ended track", "the first sequence number of the default order". 

const (
	NotSpecified  GroupSequence = 0
	FirstSequence GroupSequence = 1
	MaxSequence   GroupSequence = 0xFFFFFFFF
)

func (GroupSequence) IsInRange 

func (gs GroupSequence) IsInRange(min, max GroupSequence) bool

func (GroupSequence) Next 

func (gs GroupSequence) Next() GroupSequence

func (GroupSequence) String 

func (gs GroupSequence) String() string

type GroupWriter 

type GroupWriter interface {
	GroupSequence() GroupSequence
	WriteFrame(*Frame) error
	CloseWithError(error) error
	SetWriteDeadline(time.Time) error
	Close() error
}

* Group Writer

type Handler 

type Handler interface {
	ServeTrack(TrackWriter, SubscribeConfig)
	ServeAnnouncement(AnnouncementWriter, AnnounceConfig)
	ServeInfo(chan<- Info, InfoRequest)
}

type Info 

type Info struct {
	TrackPriority       TrackPriority
	LatestGroupSequence GroupSequence
	GroupOrder          GroupOrder
}

func (Info) String 

func (i Info) String() string

type InfoError 

type InfoError interface {
	error
	InfoErrorCode() InfoErrorCode
}

type InfoErrorCode 

type InfoErrorCode uint64

type InfoRequest 

type InfoRequest struct {
	TrackPath TrackPath
}

func (InfoRequest) String 

func (ir InfoRequest) String() string

type MediaHandler 

type MediaHandler interface {
	ServeTrack(w TrackWriter, r SubscribeConfig)
}

type ParameterType 

type ParameterType uint64

type Parameters 

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

func NewParameters 

func NewParameters() *Parameters

func (Parameters) GetBool 

func (p Parameters) GetBool(key ParameterType) (bool, error)

func (Parameters) GetByteArray 

func (p Parameters) GetByteArray(key ParameterType) ([]byte, error)

func (Parameters) GetInt 

func (p Parameters) GetInt(key ParameterType) (int64, error)

func (Parameters) GetString 

func (p Parameters) GetString(key ParameterType) (string, error)

func (Parameters) GetUint 

func (p Parameters) GetUint(key ParameterType) (uint64, error)

func (Parameters) Remove 

func (p Parameters) Remove(key ParameterType)

func (Parameters) SetBool 

func (p Parameters) SetBool(key ParameterType, value bool)

func (Parameters) SetByteArray 

func (p Parameters) SetByteArray(key ParameterType, value []byte)

func (Parameters) SetInt 

func (p Parameters) SetInt(key ParameterType, value int64)

func (Parameters) SetString 

func (p Parameters) SetString(key ParameterType, value string)

func (Parameters) SetUint 

func (p Parameters) SetUint(key ParameterType, value uint64)

func (Parameters) String 

func (p Parameters) String() string

type QUICEarlyListener 

type QUICEarlyListener interface {
	Accept(ctx context.Context) (quic.EarlyConnection, error)
	Addr() net.Addr
	Close() error
}

type ReceiveTrackStream 

type ReceiveTrackStream interface {
	TrackReader
	SubscribeID() SubscribeID
	SubscribeConfig() SubscribeConfig
	UpdateSubscribe(SubscribeUpdate) error
}

type Relayer 

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

type SendAnnounceStream 

type SendAnnounceStream interface {
	AnnouncementWriter
	AnnounceConfig() AnnounceConfig
}

type SendTrackStream 

type SendTrackStream interface {
	SubscribeID() SubscribeID
	SubscribeConfig() SubscribeConfig
	TrackWriter
}

type Server 

type Server struct {
	/*
	 * Server's Address
	 */
	Addr string

	/*
	 * TLS configuration
	 */
	TLSConfig *tls.Config

	/*
	 * QUIC configuration
	 */
	QUICConfig *quic.Config

	/*
	 * MOQ Configuration
	 */
	Config *Config

	/*
	 * Setup Extensions
	 * This function is called when a session is established
	 */
	SetupExtensions func(req Parameters) (rsp Parameters, err error)

	/*
	 * TrackMux for routing requests
	 */
	Handler Handler

	/*
	 * Session Handler
	 * This function is called when a session is established
	 */
	SessionHandlerFunc func(path string, sess Session)

	/*
	 * Logger
	 */
	Logger *slog.Logger

	/*
	 * WebTransport Server
	 * If the server is configured with a WebTransport server, it is used to handle WebTransport sessions.
	 * If not, a default server is used.
	 */
	WebtransportServer *webtransport.Server
	// contains filtered or unexported fields
}

func (*Server) Close 

func (s *Server) Close() error

func (*Server) ListenAndServe 

func (s *Server) ListenAndServe() error

func (*Server) ListenAndServeTLS 

func (s *Server) ListenAndServeTLS(certFile, keyFile string) (err error)

func (*Server) ServeWebTransport 

func (s *Server) ServeWebTransport(w http.ResponseWriter, r *http.Request) error

ServeWebTransport serves a WebTransport session. It upgrades the HTTP/3 connection to a WebTransport session and calls the session handler. If the server is not configured with a WebTransport server, it creates a default server.

func (*Server) Shutdown 

func (s *Server) Shutdown(ctx context.Context) error

type Session 

type Session interface {

	// Terminate the session
	Terminate(error)

	/*
	 * Methods for the Subscriber
	 */
	// Open an Announce Stream
	OpenAnnounceStream(AnnounceConfig) (AnnouncementReader, error)

	// Open a Track Stream
	OpenTrackStream(SubscribeConfig) (Info, ReceiveTrackStream, error)

	// Request Track Info
	RequestTrackInfo(InfoRequest) (Info, error)

	/*
	 * Methods for the Publisher
	 */
	// Accept an Announce Stream
	AcceptAnnounceStream(context.Context, func(AnnounceConfig) error) (AnnouncementWriter, error)

	// Accept a Track Stream
	AcceptTrackStream(context.Context, func(SubscribeConfig) (Info, error)) (SendTrackStream, error)

	// Accept an Info Stream
	RespondTrackInfo(context.Context, func(InfoRequest) (Info, error)) error
}

type SessionStream 

type SessionStream interface {
	UpdateSession(bitrate uint64) error
}

type SetupRequest 

type SetupRequest struct {
	Parameters Parameters
	// contains filtered or unexported fields
}

func NewSetupRequest 

func NewSetupRequest(urlstr string) (*SetupRequest, error)

func (SetupRequest) String 

func (sr SetupRequest) String() string

type SetupResponse 

type SetupResponse struct {
	Parameters Parameters
	// contains filtered or unexported fields
}

func (SetupResponse) String 

func (sr SetupResponse) String() string

type SubscribeConfig 

type SubscribeConfig struct {
	/*
	 * Required
	 */
	TrackPath TrackPath

	/*
	 * Optional
	 */
	TrackPriority    TrackPriority
	GroupOrder       GroupOrder
	MinGroupSequence GroupSequence
	MaxGroupSequence GroupSequence
}

func (SubscribeConfig) String 

func (sc SubscribeConfig) String() string

type SubscribeError 

type SubscribeError interface {
	error
	SubscribeErrorCode() SubscribeErrorCode
}

type SubscribeErrorCode 

type SubscribeErrorCode uint64

type SubscribeID 

type SubscribeID uint64

func (SubscribeID) String 

func (id SubscribeID) String() string

type SubscribeUpdate 

type SubscribeUpdate struct {
	TrackPriority    TrackPriority
	GroupOrder       GroupOrder
	MinGroupSequence GroupSequence
	MaxGroupSequence GroupSequence
}

func (SubscribeUpdate) String 

func (su SubscribeUpdate) String() string

type TerminateError 

type TerminateError interface {
	error
	TerminateErrorCode() TerminateErrorCode
}

type TerminateErrorCode 

type TerminateErrorCode uint64

type TrackBuffer 

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

func BuildTrack 

func BuildTrack(path TrackPath, info Info, expires time.Duration) *TrackBuffer

func (*TrackBuffer) Close 

func (t *TrackBuffer) Close() error

func (*TrackBuffer) CloseWithError 

func (t *TrackBuffer) CloseWithError(err error) error

func (*TrackBuffer) Count 

func (t *TrackBuffer) Count() int

func (*TrackBuffer) Info 

func (t *TrackBuffer) Info() Info

func (*TrackBuffer) LatestGroupSequence 

func (t *TrackBuffer) LatestGroupSequence() GroupSequence

func (*TrackBuffer) NewTrackReader 

func (t *TrackBuffer) NewTrackReader(priority TrackPriority, order GroupOrder) (TrackReader, error)

func (*TrackBuffer) NewTrackWriter 

func (t *TrackBuffer) NewTrackWriter(priority TrackPriority, order GroupOrder) (TrackWriter, error)

func (*TrackBuffer) ServeAnnouncement 

func (t *TrackBuffer) ServeAnnouncement(w AnnouncementWriter, config AnnounceConfig)

func (*TrackBuffer) ServeInfo 

func (t *TrackBuffer) ServeInfo(w chan<- Info, r InfoRequest)

func (*TrackBuffer) ServeTrack 

func (t *TrackBuffer) ServeTrack(w TrackWriter, config SubscribeConfig)

func (*TrackBuffer) TrackPath 

func (t *TrackBuffer) TrackPath() TrackPath

type TrackMux 

type TrackMux struct {
	// contains filtered or unexported fields
}
var DefaultMux *TrackMux = defaultMux

func NewServeMux 

func NewServeMux() *TrackMux

func (*TrackMux) Handle 

func (mux *TrackMux) Handle(pattern string, handler Handler)

func (*TrackMux) ServeAnnouncement 

func (mux *TrackMux) ServeAnnouncement(w AnnouncementWriter, r AnnounceConfig)

func (*TrackMux) ServeInfo 

func (mux *TrackMux) ServeInfo(ch chan<- Info, r InfoRequest)

func (*TrackMux) ServeTrack 

func (mux *TrackMux) ServeTrack(w TrackWriter, r SubscribeConfig)

type TrackPath 

type TrackPath string

func (TrackPath) Equal 

func (tp TrackPath) Equal(target TrackPath) bool

func (TrackPath) GetSuffix 

func (tp TrackPath) GetSuffix(prefix string) string

func (TrackPath) HasPrefix 

func (tp TrackPath) HasPrefix(prefix string) bool

func (TrackPath) HasSuffix 

func (tp TrackPath) HasSuffix(suffix string) bool

func (TrackPath) Match 

func (tp TrackPath) Match(pattern string) bool

func (TrackPath) Parts 

func (tp TrackPath) Parts() []string

func (TrackPath) String 

func (tp TrackPath) String() string

type TrackPriority 

type TrackPriority byte

type TrackReader 

type TrackReader interface {
	// Get the track path
	TrackPath() TrackPath

	// Get the track priority
	TrackPriority() TrackPriority

	// Get the group order
	GroupOrder() GroupOrder

	// Get the latest group sequence
	LatestGroupSequence() GroupSequence

	// Get the track info
	Info() Info

	// Accept a group
	AcceptGroup(context.Context) (GroupReader, error)

	Close() error

	CloseWithError(error) error
}

type TrackWriter 

type TrackWriter interface {
	// Get the track path
	TrackPath() TrackPath

	// Get the track priority
	TrackPriority() TrackPriority

	// Get the group order
	GroupOrder() GroupOrder

	// Get the latest group sequence
	LatestGroupSequence() GroupSequence

	// Get the track info
	Info() Info

	// Create a new group writer
	OpenGroup(GroupSequence) (GroupWriter, error)

	Close() error

	CloseWithError(error) error
}

Source Files

View all Source files

Directories

Path Synopsis
message
protocol
transport
transport/mock
Package mock_transport is a generated GoMock package.
 Package mock_transport is a generated GoMock package.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.