261 lines
8.8 KiB
Go
261 lines
8.8 KiB
Go
// Copyright 2023 LiveKit, Inc.
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package rpc
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/livekit/psrpc"
|
|
"github.com/livekit/psrpc/pkg/middleware"
|
|
"github.com/livekit/psrpc/pkg/middleware/otelpsrpc"
|
|
|
|
"github.com/livekit/protocol/livekit"
|
|
"github.com/livekit/protocol/logger"
|
|
"github.com/livekit/protocol/tracer"
|
|
"github.com/livekit/protocol/utils/must"
|
|
)
|
|
|
|
type PSRPCConfig struct {
|
|
MaxAttempts int `yaml:"max_attempts,omitempty"`
|
|
Timeout time.Duration `yaml:"timeout,omitempty"`
|
|
Backoff time.Duration `yaml:"backoff,omitempty"`
|
|
BufferSize int `yaml:"buffer_size,omitempty"`
|
|
}
|
|
|
|
var DefaultPSRPCConfig = PSRPCConfig{
|
|
MaxAttempts: 3,
|
|
Timeout: 3 * time.Second,
|
|
Backoff: 2 * time.Second,
|
|
BufferSize: 1000,
|
|
}
|
|
|
|
type ClientParams struct {
|
|
PSRPCConfig
|
|
Bus psrpc.MessageBus
|
|
Logger logger.Logger
|
|
TracerProvider tracer.TracerProvider // will use default if unset
|
|
Observer middleware.MetricsObserver
|
|
ClientOptions []psrpc.ClientOption
|
|
}
|
|
|
|
func NewClientParams(
|
|
config PSRPCConfig,
|
|
bus psrpc.MessageBus,
|
|
logger logger.Logger,
|
|
observer middleware.MetricsObserver,
|
|
opts ...psrpc.ClientOption,
|
|
) ClientParams {
|
|
return ClientParams{
|
|
PSRPCConfig: config,
|
|
Bus: bus,
|
|
Logger: logger,
|
|
Observer: observer,
|
|
ClientOptions: opts,
|
|
}
|
|
}
|
|
|
|
func (p *ClientParams) Options() []psrpc.ClientOption {
|
|
opts := make([]psrpc.ClientOption, 0, 4)
|
|
if p.BufferSize != 0 {
|
|
opts = append(opts, psrpc.WithClientChannelSize(p.BufferSize))
|
|
}
|
|
if p.Observer != nil {
|
|
opts = append(opts, middleware.WithClientMetrics(p.Observer))
|
|
}
|
|
if p.Logger != nil {
|
|
opts = append(opts, WithClientLogger(p.Logger))
|
|
}
|
|
if p.MaxAttempts != 0 || p.Timeout != 0 || p.Backoff != 0 {
|
|
opts = append(opts, middleware.WithRPCRetries(middleware.RetryOptions{
|
|
MaxAttempts: p.MaxAttempts,
|
|
Timeout: p.Timeout,
|
|
Backoff: p.Backoff,
|
|
}))
|
|
}
|
|
opts = append(opts, otelpsrpc.ClientOptions(otelpsrpc.Config{
|
|
TracerProvider: p.TracerProvider,
|
|
}))
|
|
opts = append(opts, p.ClientOptions...)
|
|
return opts
|
|
}
|
|
|
|
func (p *ClientParams) Args() (psrpc.MessageBus, psrpc.ClientOption) {
|
|
return p.Bus, psrpc.WithClientOptions(p.Options()...)
|
|
}
|
|
|
|
func WithServerObservability(logger logger.Logger) psrpc.ServerOption {
|
|
return psrpc.WithServerOptions(
|
|
middleware.WithServerMetrics(PSRPCMetricsObserver{}),
|
|
WithServerLogger(logger),
|
|
otelpsrpc.ServerOptions(otelpsrpc.Config{}),
|
|
)
|
|
}
|
|
|
|
func WithDefaultServerOptions(psrpcConfig PSRPCConfig, logger logger.Logger, extra ...psrpc.ServerOption) psrpc.ServerOption {
|
|
opts := []psrpc.ServerOption{
|
|
psrpc.WithServerChannelSize(psrpcConfig.BufferSize),
|
|
WithServerObservability(logger),
|
|
}
|
|
opts = append(opts, extra...)
|
|
return psrpc.WithServerOptions(opts...)
|
|
}
|
|
|
|
func WithClientObservability(logger logger.Logger) psrpc.ClientOption {
|
|
return psrpc.WithClientOptions(
|
|
middleware.WithClientMetrics(PSRPCMetricsObserver{}),
|
|
WithClientLogger(logger),
|
|
otelpsrpc.ClientOptions(otelpsrpc.Config{}),
|
|
)
|
|
}
|
|
|
|
func WithDefaultClientOptions(logger logger.Logger, extra ...psrpc.ClientOption) psrpc.ClientOption {
|
|
opts := []psrpc.ClientOption{
|
|
WithClientObservability(logger),
|
|
}
|
|
opts = append(opts, extra...)
|
|
return psrpc.WithClientOptions(opts...)
|
|
}
|
|
|
|
func WithPropagateRequestTimeout(ctx context.Context) psrpc.RequestOption {
|
|
if deadline, ok := ctx.Deadline(); ok {
|
|
return psrpc.WithRequestTimeout(time.Until(deadline))
|
|
}
|
|
return func(ro *psrpc.RequestOpts) {}
|
|
}
|
|
|
|
//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate
|
|
|
|
type TypedSignalClient = SignalClient[livekit.NodeID]
|
|
type TypedSignalServer = SignalServer[livekit.NodeID]
|
|
|
|
func NewTypedSignalClient(nodeID livekit.NodeID, bus psrpc.MessageBus, opts ...psrpc.ClientOption) (TypedSignalClient, error) {
|
|
return NewSignalClient[livekit.NodeID](bus, psrpc.WithClientOptions(opts...), psrpc.WithClientID(string(nodeID)))
|
|
}
|
|
|
|
func NewTypedSignalServer(nodeID livekit.NodeID, svc SignalServerImpl, bus psrpc.MessageBus, opts ...psrpc.ServerOption) (TypedSignalServer, error) {
|
|
return NewSignalServer[livekit.NodeID](svc, bus, psrpc.WithServerOptions(opts...), psrpc.WithServerID(string(nodeID)))
|
|
}
|
|
|
|
type TypedRoomManagerClient = RoomManagerClient[livekit.NodeID]
|
|
type TypedRoomManagerServer = RoomManagerServer[livekit.NodeID]
|
|
|
|
func NewTypedRoomManagerClient(bus psrpc.MessageBus, opts ...psrpc.ClientOption) (TypedRoomManagerClient, error) {
|
|
return NewRoomManagerClient[livekit.NodeID](bus, opts...)
|
|
}
|
|
|
|
func NewTypedRoomManagerServer(svc RoomManagerServerImpl, bus psrpc.MessageBus, opts ...psrpc.ServerOption) (TypedRoomManagerServer, error) {
|
|
return NewRoomManagerServer[livekit.NodeID](svc, bus, opts...)
|
|
}
|
|
|
|
type ParticipantTopic string
|
|
type RoomTopic string
|
|
|
|
func FormatParticipantTopic(roomName livekit.RoomName, identity livekit.ParticipantIdentity) ParticipantTopic {
|
|
return ParticipantTopic(fmt.Sprintf("%s_%s", roomName, identity))
|
|
}
|
|
|
|
func FormatRoomTopic(roomName livekit.RoomName) RoomTopic {
|
|
return RoomTopic(roomName)
|
|
}
|
|
|
|
type topicFormatter struct{}
|
|
|
|
func NewTopicFormatter() TopicFormatter {
|
|
return topicFormatter{}
|
|
}
|
|
|
|
func (f topicFormatter) ParticipantTopic(ctx context.Context, roomName livekit.RoomName, identity livekit.ParticipantIdentity) ParticipantTopic {
|
|
return FormatParticipantTopic(roomName, identity)
|
|
}
|
|
|
|
func (f topicFormatter) RoomTopic(ctx context.Context, roomName livekit.RoomName) RoomTopic {
|
|
return FormatRoomTopic(roomName)
|
|
}
|
|
|
|
type TopicFormatter interface {
|
|
ParticipantTopic(ctx context.Context, roomName livekit.RoomName, identity livekit.ParticipantIdentity) ParticipantTopic
|
|
RoomTopic(ctx context.Context, roomName livekit.RoomName) RoomTopic
|
|
}
|
|
|
|
//counterfeiter:generate . TypedRoomClient
|
|
type TypedRoomClient = RoomClient[RoomTopic]
|
|
type TypedRoomServer = RoomServer[RoomTopic]
|
|
|
|
func NewTypedRoomClient(params ClientParams) (TypedRoomClient, error) {
|
|
return NewRoomClient[RoomTopic](params.Args())
|
|
}
|
|
|
|
func NewTypedRoomServer(svc RoomServerImpl, bus psrpc.MessageBus, opts ...psrpc.ServerOption) (TypedRoomServer, error) {
|
|
return NewRoomServer[RoomTopic](svc, bus, opts...)
|
|
}
|
|
|
|
//counterfeiter:generate . TypedParticipantClient
|
|
type TypedParticipantClient = ParticipantClient[ParticipantTopic]
|
|
type TypedParticipantServer = ParticipantServer[ParticipantTopic]
|
|
|
|
func NewTypedParticipantClient(params ClientParams) (TypedParticipantClient, error) {
|
|
return NewParticipantClient[ParticipantTopic](params.Args())
|
|
}
|
|
|
|
func NewTypedParticipantServer(svc ParticipantServerImpl, bus psrpc.MessageBus, opts ...psrpc.ServerOption) (TypedParticipantServer, error) {
|
|
return NewParticipantServer[ParticipantTopic](svc, bus, opts...)
|
|
}
|
|
|
|
//counterfeiter:generate . TypedWHIPParticipantClient
|
|
type TypedWHIPParticipantClient = WHIPParticipantClient[ParticipantTopic]
|
|
type TypedWHIPParticipantServer = WHIPParticipantServer[ParticipantTopic]
|
|
|
|
func NewTypedWHIPParticipantClient(params ClientParams) (TypedWHIPParticipantClient, error) {
|
|
return NewWHIPParticipantClient[ParticipantTopic](params.Args())
|
|
}
|
|
|
|
func NewTypedWHIPParticipantServer(svc WHIPParticipantServerImpl, bus psrpc.MessageBus, opts ...psrpc.ServerOption) (TypedWHIPParticipantServer, error) {
|
|
return NewWHIPParticipantServer[ParticipantTopic](svc, bus, opts...)
|
|
}
|
|
|
|
//counterfeiter:generate . TypedAgentDispatchInternalClient
|
|
type TypedAgentDispatchInternalClient = AgentDispatchInternalClient[RoomTopic]
|
|
type TypedAgentDispatchInternalServer = AgentDispatchInternalServer[RoomTopic]
|
|
|
|
func NewTypedAgentDispatchInternalClient(params ClientParams) (TypedAgentDispatchInternalClient, error) {
|
|
return NewAgentDispatchInternalClient[RoomTopic](params.Args())
|
|
}
|
|
|
|
func NewTypedAgentDispatchInternalServer(svc AgentDispatchInternalServerImpl, bus psrpc.MessageBus, opts ...psrpc.ServerOption) (TypedAgentDispatchInternalServer, error) {
|
|
return NewAgentDispatchInternalServer[RoomTopic](svc, bus, opts...)
|
|
}
|
|
|
|
//counterfeiter:generate . KeepalivePubSub
|
|
type KeepalivePubSub interface {
|
|
KeepaliveClient[livekit.NodeID]
|
|
KeepaliveServer[livekit.NodeID]
|
|
}
|
|
|
|
type keepalivePubSub struct {
|
|
KeepaliveClient[livekit.NodeID]
|
|
KeepaliveServer[livekit.NodeID]
|
|
}
|
|
|
|
func NewKeepalivePubSub(params ClientParams) (KeepalivePubSub, error) {
|
|
client, err := NewKeepaliveClient[livekit.NodeID](params.Args())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
server := must.Get(NewKeepaliveServer[livekit.NodeID](nil, params.Bus))
|
|
return &keepalivePubSub{client, server}, nil
|
|
}
|