broker

Documentation

Overview

Package broker implements the actual webrtc broker

Index

• type Broker
  • func NewBroker(config *Config) (*Broker, error)
  • func (b *Broker) Connect() error
  • func (b *Broker) GenerateCoordinatorConnectURL() (string, error)
  • func (b *Broker) GetBrokerStats() Stats
  • func (b *Broker) ProcessMessagesChannel()
  • func (b *Broker) ProcessSubscriptionChannel()
  • func (b *Broker) Shutdown()
• type BufferedWriterController
  • func NewBufferedWriterController(writer PeerWriter, initialQueueSize int, maxBufferSize uint64) *BufferedWriterController
  • func (c *BufferedWriterController) OnBufferedAmountLow()
  • func (c *BufferedWriterController) Write(p []byte)
• type Config
• type DiscardWriterController
  • func NewDiscardWriterController(writer PeerWriter, maxBufferSize uint64) *DiscardWriterController
  • func (c *DiscardWriterController) GetDiscardedCount() uint32
  • func (c *DiscardWriterController) OnBufferedAmountLow()
  • func (c *DiscardWriterController) Write(p []byte)
• type FixedQueueWriterController
  • func NewFixedQueueWriterController(writer PeerWriter, queueSize int, maxBufferSize uint64) *FixedQueueWriterController
  • func (c *FixedQueueWriterController) GetDiscardedCount() uint32
  • func (c *FixedQueueWriterController) OnBufferedAmountLow()
  • func (c *FixedQueueWriterController) Write(p []byte)
• type GzipCompression
  • func (g *GzipCompression) Unzip(zipped []byte) ([]byte, error)
  • func (g *GzipCompression) Zip(plain []byte) ([]byte, error)
• type PeerStats
• type PeerWriter
• type Stats
• type StatsSummary
• type StatsSummaryGenerator
  • func NewStatsSummaryGenerator() StatsSummaryGenerator
  • func (g *StatsSummaryGenerator) Generate(stats Stats) StatsSummary
• type UnboundedWriterController
  • func NewUnboundedWriterController(writer PeerWriter) *UnboundedWriterController
  • func (c *UnboundedWriterController) OnBufferedAmountLow()
  • func (c *UnboundedWriterController) Write(p []byte)
• type WriterController
• type WriterControllerFactory
• type ZipCompression

Types

type Broker

type Broker struct {
	*server.Server
	// contains filtered or unexported fields
}

Broker ...

func NewBroker

func NewBroker(config *Config) (*Broker, error)

NewBroker creates a new broker

func (*Broker) Connect

func (b *Broker) Connect() error

Connect connects the broker to the coordinator

func (*Broker) GenerateCoordinatorConnectURL

func (b *Broker) GenerateCoordinatorConnectURL() (string, error)

GenerateCoordinatorConnectURL ...

func (*Broker) GetBrokerStats

func (b *Broker) GetBrokerStats() Stats

GetBrokerStats ...

func (*Broker) ProcessMessagesChannel

func (b *Broker) ProcessMessagesChannel()

ProcessMessagesChannel start the topic message processor

func (*Broker) ProcessSubscriptionChannel

func (b *Broker) ProcessSubscriptionChannel()

ProcessSubscriptionChannel start the subscription message processor

func (*Broker) Shutdown

func (b *Broker) Shutdown()

Shutdown ...

type BufferedWriterController

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

BufferedWriterController provides an ever growing queue between the writer and the user

func NewBufferedWriterController

func NewBufferedWriterController(
	writer PeerWriter, initialQueueSize int, maxBufferSize uint64) *BufferedWriterController

NewBufferedWriterController creates a new BufferedWriterController

func (*BufferedWriterController) OnBufferedAmountLow

func (c *BufferedWriterController) OnBufferedAmountLow()

OnBufferedAmountLow ...

func (*BufferedWriterController) Write

func (c *BufferedWriterController) Write(p []byte)

Write ...

type Config

type Config struct {
	CoordinatorURL                              string
	Log                                         *logging.Logger
	ICEServers                                  []pion.ICEServer
	Zipper                                      ZipCompression
	Auth                                        authentication.ServerAuthenticator
	ReliableWriterControllerFactory             WriterControllerFactory
	UnreliableWriterControllerFactory           WriterControllerFactory
	ReliableChannelBufferedAmountLowThreshold   uint64
	UnreliableChannelBufferedAmountLowThreshold uint64

	MaxPeers               uint16
	ExitOnCoordinatorClose bool
	WebRtcLogLevel         zerolog.Level
	Role                   protocol.Role
}

Config is the broker config

type DiscardWriterController

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

DiscardWriterController simply discard any packages when the BufferedAmount > maxBufferSize

func NewDiscardWriterController

func NewDiscardWriterController(writer PeerWriter, maxBufferSize uint64) *DiscardWriterController

NewDiscardWriterController creates a new DiscardWriterController

func (*DiscardWriterController) GetDiscardedCount

func (c *DiscardWriterController) GetDiscardedCount() uint32

GetDiscardedCount ...

func (*DiscardWriterController) OnBufferedAmountLow

func (c *DiscardWriterController) OnBufferedAmountLow()

OnBufferedAmountLow ...

func (*DiscardWriterController) Write

func (c *DiscardWriterController) Write(p []byte)

Write ...

type FixedQueueWriterController

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

FixedQueueWriterController simply discard any packages when the BufferedAmount > maxBufferSize

func NewFixedQueueWriterController

func NewFixedQueueWriterController(writer PeerWriter, queueSize int, maxBufferSize uint64) *FixedQueueWriterController

NewFixedQueueWriterController creates a new FixedQueueWriterController

func (*FixedQueueWriterController) GetDiscardedCount

func (c *FixedQueueWriterController) GetDiscardedCount() uint32

GetDiscardedCount ...

func (*FixedQueueWriterController) OnBufferedAmountLow

func (c *FixedQueueWriterController) OnBufferedAmountLow()

OnBufferedAmountLow ...

func (*FixedQueueWriterController) Write

func (c *FixedQueueWriterController) Write(p []byte)

Write ...

type GzipCompression

type GzipCompression struct{}

GzipCompression compressor for gzip format

func (*GzipCompression) Unzip

func (g *GzipCompression) Unzip(zipped []byte) ([]byte, error)

Unzip the given byte array

func (*GzipCompression) Zip

func (g *GzipCompression) Zip(plain []byte) ([]byte, error)

Zip the given byte array

type PeerStats

type PeerStats struct {
	Alias      uint64
	Identity   []byte
	State      pion.ICEConnectionState
	TopicCount uint32

	Nomination          bool
	LocalCandidateType  pion.ICECandidateType
	RemoteCandidateType pion.ICECandidateType

	DataChannelsOpened    uint32
	DataChannelsClosed    uint32
	DataChannelsRequested uint32
	DataChannelsAccepted  uint32

	ReliableProtocol         string
	ReliableState            pion.DataChannelState
	ReliableBytesSent        uint64
	ReliableBytesReceived    uint64
	ReliableMessagesSent     uint32
	ReliableMessagesReceived uint32
	ReliableBufferedAmount   uint64

	UnreliableProtocol         string
	UnreliableState            pion.DataChannelState
	UnreliableBytesSent        uint64
	UnreliableBytesReceived    uint64
	UnreliableMessagesSent     uint32
	UnreliableMessagesReceived uint32
	UnreliableBufferedAmount   uint64

	ICETransportBytesSent     uint64
	ICETransportBytesReceived uint64

	SCTPTransportBytesSent     uint64
	SCTPTransportBytesReceived uint64
}

PeerStats ...

type PeerWriter

type PeerWriter interface {
	BufferedAmount() uint64
	Write(p []byte) error
}

PeerWriter represents the basic write operations for a peer dc

type Stats

type Stats struct {
	Time                time.Time
	Alias               uint64
	ConnectChSize       int
	WebRtcControlChSize int
	UnregisterChSize    int
	TopicCount          int
	SubscriptionChSize  int
	MessagesChSize      int
	Peers               map[uint64]PeerStats
}

Stats ...

type StatsSummary

type StatsSummary struct {
	From time.Time
	To   time.Time

	StateCount               map[pion.ICEConnectionState]uint32
	LocalCandidateTypeCount  map[pion.ICECandidateType]uint32
	RemoteCandidateTypeCount map[pion.ICECandidateType]uint32

	MessagesSentByDC, BytesSentByDC, BytesSentByICE, BytesSentBySCTP                 uint64
	MessagesReceivedByDC, BytesReceivedByDC, BytesReceivedByICE, BytesReceivedBySCTP uint64
}

StatsSummary ...

type StatsSummaryGenerator

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

StatsSummaryGenerator ...

func NewStatsSummaryGenerator

func NewStatsSummaryGenerator() StatsSummaryGenerator

NewStatsSummaryGenerator creates a StatsSummaryGenerator

func (*StatsSummaryGenerator) Generate

func (g *StatsSummaryGenerator) Generate(stats Stats) StatsSummary

Generate generates a new StatsSummary comparing the new stats with the last processed one

type UnboundedWriterController

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

UnboundedWriterController simply discard any packages when the BufferedAmount > maxBufferSize

func NewUnboundedWriterController

func NewUnboundedWriterController(writer PeerWriter) *UnboundedWriterController

NewUnboundedWriterController creates a new UnboundedWriterController

func (*UnboundedWriterController) OnBufferedAmountLow

func (c *UnboundedWriterController) OnBufferedAmountLow()

OnBufferedAmountLow ...

func (*UnboundedWriterController) Write

func (c *UnboundedWriterController) Write(p []byte)

Write ...

type WriterController

type WriterController interface {
	Write(byte []byte)
	OnBufferedAmountLow()
}

WriterController is in charge of the peer writer flow control

type WriterControllerFactory

type WriterControllerFactory = func(uint64, PeerWriter) WriterController

WriterControllerFactory ...

type ZipCompression

type ZipCompression interface {
	Zip(plain []byte) ([]byte, error)
	Unzip(zipped []byte) ([]byte, error)
}

ZipCompression is the zip compression interface