mirror of
https://github.com/livekit/livekit.git
synced 2026-08-29 05:29:32 +00:00
Merge branch 'master' into ninad/latency-metric-track-status-too-01
This commit is contained in:
Ninad Pundalik
@@ -0,0 +1,53 @@
|
||||
# Copyright 2026 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.
|
||||
|
||||
# Builds and publishes the SDK conformance test server (cmd/test-server) to
|
||||
# Docker Hub as livekit/test-server:latest on every push to master. The server
|
||||
# SDK repos boot this image in their CI to run region-failover tests.
|
||||
name: Release Test Server to Docker
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
on:
|
||||
workflow_dispatch:
|
||||
push:
|
||||
branches:
|
||||
- master
|
||||
|
||||
jobs:
|
||||
docker:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6
|
||||
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd # v4
|
||||
|
||||
- name: Login to DockerHub
|
||||
uses: docker/login-action@650006c6eb7dba73a995cc03b0b2d7f5ca915bee # v4
|
||||
with:
|
||||
username: ${{ secrets.DOCKERHUB_USERNAME }}
|
||||
password: ${{ secrets.DOCKERHUB_TOKEN }}
|
||||
|
||||
- name: Build and push
|
||||
uses: docker/build-push-action@f9f3042f7e2789586610d6e8b85c8f03e5195baf # v7
|
||||
with:
|
||||
context: .
|
||||
file: cmd/test-server/Dockerfile
|
||||
push: true
|
||||
platforms: linux/amd64,linux/arm64
|
||||
# Each build overrides the latest tag so SDK CI always boots the
|
||||
# current mock server.
|
||||
tags: livekit/test-server:latest
|
||||
@@ -2,6 +2,58 @@
|
||||
|
||||
This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
|
||||
|
||||
## [1.13.3] - 2026-07-03
|
||||
|
||||
### Added
|
||||
|
||||
- feat: mock API server for testing server SDKs (#4627)
|
||||
- support auth checks with mock server (#4629)
|
||||
- Data track schema metadata (#4622)
|
||||
- Report average bitrates for whip ingress (#4634)
|
||||
|
||||
### Changed
|
||||
|
||||
- Update webrtc to fix interop issue with bundled datachannel (#4631)
|
||||
- Update module github.com/urfave/cli/v3 to v3.10.0 (#4612)
|
||||
- Use camel case log name in `DataBlobKey` (#4633)
|
||||
|
||||
### Fixed
|
||||
|
||||
- Stop WHIP session notifier when participant leaves (#4637)
|
||||
|
||||
## [1.13.2] - 2026-06-27
|
||||
|
||||
### Added
|
||||
|
||||
- Add Prometheus metrics for join latency and peer connection state (#4574, #4616)
|
||||
- Preserve original expiry when refreshing token (#4580)
|
||||
- Add grants expiry to Auth context (#4581)
|
||||
- Add ability to run pprof on dedicated HTTP server (#4584)
|
||||
- Add API to get latest node stats (#4589)
|
||||
- Enforce subscription permission to data track (#4588)
|
||||
- egress v2 api (#4592)
|
||||
- agent: thread attributes map from dispatch to job (#4598)
|
||||
- Log subscription limit breaches (#4603)
|
||||
- Acquire requested video layer directly at HIGH quality by default (#4595)
|
||||
- Report participant capabilities in ParticipantInfo (#4606)
|
||||
- Add option to force drain rtcService/agentService connections (#4618)
|
||||
- Add support for data blob (a. k. a. async participant attributes) (#4619)
|
||||
|
||||
### Changed
|
||||
|
||||
- Update dependencies: pion/sctp, DTLS v3.1.4, protocol (#4587, #4601, #4623)
|
||||
- rtc: add RestartSessionTimer to re-anchor participant session duration (#4566)
|
||||
- Record more RTC cancellation points (#4600)
|
||||
- Cap all metadata at 512 KiB; enforce on join, agent dispatch, and embedded agents (#4602)
|
||||
- Tighten up publish latency stat (#4615)
|
||||
- Echo offered audio payload types in single-PC subscriber answer (#4614)
|
||||
|
||||
### Fixed
|
||||
|
||||
- Fix skipped packets accounting (#4604)
|
||||
- Do not log due to negative getting interpreted as large unsigned positive (#4605)
|
||||
- Do not call nil callback (#4607)
|
||||
|
||||
## [1.13.1] - 2026-06-08
|
||||
|
||||
### Fixed
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
# Copyright 2026 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.
|
||||
|
||||
# Mock LiveKit API server for SDK conformance testing. Build from the repo
|
||||
# root: docker build -f cmd/test-server/Dockerfile -t livekit/test-server .
|
||||
FROM golang:1.26-alpine AS builder
|
||||
|
||||
ARG TARGETARCH
|
||||
|
||||
WORKDIR /workspace
|
||||
|
||||
COPY go.mod go.mod
|
||||
COPY go.sum go.sum
|
||||
RUN go mod download
|
||||
|
||||
COPY cmd/ cmd/
|
||||
COPY pkg/ pkg/
|
||||
COPY version/ version/
|
||||
|
||||
RUN CGO_ENABLED=0 GOOS=linux GOARCH=$TARGETARCH GO111MODULE=on go build -a -o livekit-test-server ./cmd/test-server
|
||||
|
||||
FROM alpine
|
||||
|
||||
COPY --from=builder /workspace/livekit-test-server /livekit-test-server
|
||||
|
||||
# region-0 (primary, 9999) + 3 fallback regions
|
||||
EXPOSE 9999 10000 10001 10002
|
||||
|
||||
ENTRYPOINT ["/livekit-test-server"]
|
||||
@@ -0,0 +1,138 @@
|
||||
# LiveKit SDK test server
|
||||
|
||||
A stateless, per-request programmable mock of the LiveKit server HTTP API. It
|
||||
exists so the server SDKs (Go, Rust, Python, Node, Kotlin, Ruby) can exercise
|
||||
client-side behavior against one shared implementation, published as a Docker
|
||||
image and booted by each SDK's CI.
|
||||
|
||||
## Why it looks the way it does
|
||||
|
||||
- **Stateless.** All behavior is selected by a single per-request `X-Lk-Mock`
|
||||
header (a JSON object), so the server holds no mutable state and tests run in
|
||||
parallel.
|
||||
- **Multi-port = multi-region.** The process binds one listener per simulated
|
||||
region (`--ports`). A port's position in the list is its **region index**;
|
||||
index `0` is the primary the SDK is initially pointed at. `GET
|
||||
/settings/regions` advertises all of them in order.
|
||||
- **One header drives every attempt.** The SDK sends the same control header on
|
||||
the initial request *and* every failover retry. Each listener decides what to
|
||||
do from its **own** index, so a single `X-Lk-Mock: {"failRegions":[0]}` makes
|
||||
the primary fail while the first fallback succeeds — no coordination needed.
|
||||
- **Realistic latency.** Methods that block in the real server block here too:
|
||||
`CreateSIPParticipant` with `wait_until_answered` and `TransferSIPParticipant`
|
||||
take ~11s before responding, so SDKs can exercise their timeouts.
|
||||
- **The whole API is mocked with populated responses.** Every RoomService,
|
||||
Egress, Ingress, SIP, and Connector method returns a type-correct, populated
|
||||
response: scalar fields that share a name with the request are echoed (e.g.
|
||||
`name`, `metadata`, `identity`, timeouts), `id`/`sid` fields get placeholder
|
||||
values, and list endpoints return one element. Both protobuf and JSON Twirp
|
||||
clients are supported. A client can override the response entirely with the
|
||||
`response` field (see below). Unregistered/future methods fall back
|
||||
to an empty (all-default) message, which still decodes cleanly.
|
||||
|
||||
## Running
|
||||
|
||||
```bash
|
||||
go run ./cmd/test-server # primary :9999, regions :10000-10002
|
||||
go run ./cmd/test-server --ports 9999,10000 # primary + one fallback
|
||||
|
||||
# Docker
|
||||
docker build -f cmd/test-server/Dockerfile -t livekit/test-server .
|
||||
docker run -p 9999-10002:9999-10002 livekit/test-server
|
||||
```
|
||||
|
||||
| Flag | Env | Default | Meaning |
|
||||
|---|---|---|---|
|
||||
| `--ports` | `LK_TEST_SERVER_PORTS` | `9999,10000,10001,10002` | listener ports; index = position |
|
||||
| `--advertise-host` | `LK_TEST_SERVER_ADVERTISE_HOST` | `http://127.0.0.1` | base URL used in `/settings/regions` |
|
||||
| `--bind` | `LK_TEST_SERVER_BIND` | `0.0.0.0` | bind address |
|
||||
| `--twirp-prefix` | `LK_TEST_SERVER_TWIRP_PREFIX` | `/twirp` | Twirp path prefix |
|
||||
|
||||
## Control protocol
|
||||
|
||||
All behavior is driven by a single `X-Lk-Mock` request header whose value is a
|
||||
JSON object. The SDK sends the same header on API calls, on the
|
||||
`/settings/regions` fetch, and on every failover retry (it must forward
|
||||
client-configured custom headers onto all of them). Omit the header — or any
|
||||
field — for normal behavior. Every field is optional:
|
||||
|
||||
| Field | Default | Effect |
|
||||
|---|---|---|
|
||||
| `failRegions` | — | array of region indices that fail this request, e.g. `[0]` or `[0,1]`. Each listener fails only if its own index is listed. |
|
||||
| `failMode` | `status` | how a failing region fails: `status` (write a Twirp error), or `drop` (close the connection → transport error). |
|
||||
| `failStatus` | `503` | HTTP status for a `status`-mode failure. |
|
||||
| `failTwirpCode` | derived from status | Twirp error code string in the failure body. |
|
||||
| `delayMs` | — | delay (ms) before responding, on success or failure. Overrides a method's natural latency — use it for timeout tests, or set it to skip a SIP method's built-in ~11s wait. |
|
||||
| `regionsStatus` | `200` | override the status of `GET /settings/regions`. |
|
||||
| `response` | — | the response message for the called method (a JSON object, protojson-shaped); replaces the populated default, giving full control over the returned payload. |
|
||||
| `skipAuth` | `false` | `true` disables permission enforcement for the request (use for tests that aren't about authz, e.g. failover tests with a placeholder token). |
|
||||
| `sipStatus` | — | fail a SIP dial method (`CreateSIPParticipant`/`TransferSIPParticipant`) with a SIP status, e.g. `{"code":486,"status":"Busy Here"}` (`status` optional). The Twirp error code and `sip_status_code`/`sip_status`/`error_details` metadata are derived from it exactly as the real server does. Composes with `delayMs` to simulate "ring, then fail". |
|
||||
|
||||
Example: `X-Lk-Mock: {"skipAuth":true,"failRegions":[0],"failStatus":400}`
|
||||
|
||||
> **Deprecated:** the older per-setting headers — `X-Lk-Mock-Fail-Regions`,
|
||||
> `X-Lk-Mock-Fail-Mode` (incl. the `delay` mode), `X-Lk-Mock-Fail-Status`,
|
||||
> `X-Lk-Mock-Fail-Twirp-Code`, `X-Lk-Mock-Delay-Ms`, `X-Lk-Mock-Regions-Status`,
|
||||
> `X-Lk-Mock-Response`, `X-Lk-Mock-Skip-Auth` — are still honored for existing
|
||||
> clients and will be removed later. When `X-Lk-Mock` is also present, its fields
|
||||
> take precedence per-field. New clients should use `X-Lk-Mock` only.
|
||||
|
||||
Response headers:
|
||||
|
||||
| Header | Meaning |
|
||||
|---|---|
|
||||
| `X-Lk-Mock-Region` | index of the region that served the response (blank on a failed region). Assert on this to confirm which region a failover landed on. |
|
||||
|
||||
## Permission enforcement
|
||||
|
||||
Every API method requires the same token grants the real LiveKit server checks
|
||||
(see `pkg/service/auth.go`), so the mock doubles as a conformance check that an
|
||||
SDK attaches the right permissions automatically. Tokens are parsed and verified
|
||||
with the protocol's own `auth` helpers — the same code path the real server uses
|
||||
— against the mock's configured API secret (default `secret`, matching
|
||||
`livekit-server --dev`; override with `--api-secret` / `LK_TEST_SERVER_API_SECRET`).
|
||||
|
||||
- Missing, malformed, or wrongly-signed `Authorization` → `401 unauthenticated`.
|
||||
- Validly-signed token without the required grant → `403 permission_denied`.
|
||||
- `roomAdmin`-scoped methods also require the token's `room` to match the
|
||||
request's room; `ForwardParticipant`/`MoveParticipant` additionally require
|
||||
`destinationRoom` to match.
|
||||
|
||||
SDKs exercising permissions should sign tokens with the same API secret the mock
|
||||
is configured with (`secret` by default).
|
||||
|
||||
| Grant | Methods |
|
||||
|---|---|
|
||||
| `video.roomCreate` | `CreateRoom`, `DeleteRoom`, all `Connector` calls |
|
||||
| `video.roomList` | `ListRooms` |
|
||||
| `video.roomRecord` | all `Egress` methods |
|
||||
| `video.ingressAdmin` | all `Ingress` methods |
|
||||
| `video.roomAdmin` (+ `room`) | room participant/data/metadata methods, `AgentDispatchService` methods |
|
||||
| `video.roomAdmin` (+ `room` + `destinationRoom`) | `ForwardParticipant`, `MoveParticipant` |
|
||||
| `sip.admin` | SIP trunk & dispatch-rule CRUD |
|
||||
| `sip.call` | `CreateSIPParticipant`; `TransferSIPParticipant` (also needs `roomAdmin`) |
|
||||
|
||||
Send `X-Lk-Mock: {"skipAuth":true}` to bypass enforcement for tests that aren't
|
||||
about permissions.
|
||||
|
||||
## Common recipes
|
||||
|
||||
| Goal | `X-Lk-Mock` value |
|
||||
|---|---|
|
||||
| Happy path | (no header) — valid token with the method's grant → 200 from region `0` |
|
||||
| Bypass auth (failover tests) | `{"skipAuth":true}` |
|
||||
| Missing-permission error | (no header) — token without the required grant → 403 |
|
||||
| Failover succeeds on region 1 | `{"failRegions":[0]}` |
|
||||
| Exhaust to region 2 | `{"failRegions":[0,1]}` |
|
||||
| All regions down | `{"failRegions":[0,1,2,3]}` |
|
||||
| 4xx, no retry | `{"failRegions":[0],"failStatus":400}` |
|
||||
| Transport-error failover | `{"failRegions":[0],"failMode":"drop"}` |
|
||||
| Timeout test | `{"delayMs":30000}` |
|
||||
| Region discovery unreachable | `{"regionsStatus":500}` |
|
||||
| Custom response payload | `{"response":{"sid":"RM_x","name":"my-room"}}` |
|
||||
| SIP busy signal | `{"sipStatus":{"code":486,"status":"Busy Here"}}` |
|
||||
| SIP carrier decline | `{"sipStatus":{"code":603}}` |
|
||||
|
||||
Note: SDK region failover normally only engages for `*.livekit.cloud` hosts.
|
||||
Since tests point at `127.0.0.1`, set the SDK's failover-enable option to its
|
||||
forced-on value so failover engages against localhost.
|
||||
@@ -0,0 +1,224 @@
|
||||
// Copyright 2026 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 main
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"github.com/livekit/protocol/auth"
|
||||
|
||||
"google.golang.org/protobuf/proto"
|
||||
"google.golang.org/protobuf/reflect/protoreflect"
|
||||
)
|
||||
|
||||
// The mock enforces the same token permissions the real LiveKit server requires
|
||||
// for each API method (see pkg/service/auth.go in livekit/livekit). The point is
|
||||
// to verify that SDKs attach the correct grants automatically. Tokens are parsed
|
||||
// and verified with the protocol's own auth helpers (the same code the server
|
||||
// uses), against the mock's configured API secret (default "secret", matching
|
||||
// `livekit-server --dev`); set --api-secret / LK_TEST_SERVER_API_SECRET to change it.
|
||||
//
|
||||
// Set `"skipAuth": true` in the X-Lk-Mock header to bypass enforcement for tests
|
||||
// that aren't about permissions (e.g. region-failover tests with a placeholder token).
|
||||
|
||||
// perm describes the grants a method requires. roomAdmin additionally requires
|
||||
// the token's room to match the request's room; destRoom further requires the
|
||||
// token's destinationRoom to match the request's destination_room.
|
||||
type perm struct {
|
||||
roomCreate bool
|
||||
roomList bool
|
||||
roomRecord bool
|
||||
ingressAdmin bool
|
||||
roomAdmin bool
|
||||
destRoom bool
|
||||
sipAdmin bool
|
||||
sipCall bool
|
||||
}
|
||||
|
||||
// methodPerms maps "<package>.<Service>/<Method>" to its required grants,
|
||||
// matching the Ensure*Permission checks in the real server's services.
|
||||
var methodPerms = map[string]perm{
|
||||
// RoomService
|
||||
"livekit.RoomService/CreateRoom": {roomCreate: true},
|
||||
"livekit.RoomService/DeleteRoom": {roomCreate: true},
|
||||
"livekit.RoomService/ListRooms": {roomList: true},
|
||||
"livekit.RoomService/ListParticipants": {roomAdmin: true},
|
||||
"livekit.RoomService/GetParticipant": {roomAdmin: true},
|
||||
"livekit.RoomService/RemoveParticipant": {roomAdmin: true},
|
||||
"livekit.RoomService/MutePublishedTrack": {roomAdmin: true},
|
||||
"livekit.RoomService/UpdateParticipant": {roomAdmin: true},
|
||||
"livekit.RoomService/UpdateSubscriptions": {roomAdmin: true},
|
||||
"livekit.RoomService/SendData": {roomAdmin: true},
|
||||
"livekit.RoomService/UpdateRoomMetadata": {roomAdmin: true},
|
||||
"livekit.RoomService/ForwardParticipant": {destRoom: true},
|
||||
"livekit.RoomService/MoveParticipant": {destRoom: true},
|
||||
"livekit.RoomService/PerformRpc": {roomAdmin: true},
|
||||
|
||||
// Egress — all require record permission
|
||||
"livekit.Egress/StartEgress": {roomRecord: true},
|
||||
"livekit.Egress/StartRoomCompositeEgress": {roomRecord: true},
|
||||
"livekit.Egress/StartWebEgress": {roomRecord: true},
|
||||
"livekit.Egress/StartParticipantEgress": {roomRecord: true},
|
||||
"livekit.Egress/StartTrackCompositeEgress": {roomRecord: true},
|
||||
"livekit.Egress/StartTrackEgress": {roomRecord: true},
|
||||
"livekit.Egress/UpdateLayout": {roomRecord: true},
|
||||
"livekit.Egress/UpdateStream": {roomRecord: true},
|
||||
"livekit.Egress/ListEgress": {roomRecord: true},
|
||||
"livekit.Egress/StopEgress": {roomRecord: true},
|
||||
|
||||
// Ingress — all require ingress admin
|
||||
"livekit.Ingress/CreateIngress": {ingressAdmin: true},
|
||||
"livekit.Ingress/UpdateIngress": {ingressAdmin: true},
|
||||
"livekit.Ingress/ListIngress": {ingressAdmin: true},
|
||||
"livekit.Ingress/DeleteIngress": {ingressAdmin: true},
|
||||
|
||||
// SIP — trunk/dispatch administration requires sip.admin
|
||||
"livekit.SIP/CreateSIPInboundTrunk": {sipAdmin: true},
|
||||
"livekit.SIP/CreateSIPOutboundTrunk": {sipAdmin: true},
|
||||
"livekit.SIP/UpdateSIPInboundTrunk": {sipAdmin: true},
|
||||
"livekit.SIP/UpdateSIPOutboundTrunk": {sipAdmin: true},
|
||||
"livekit.SIP/GetSIPInboundTrunk": {sipAdmin: true},
|
||||
"livekit.SIP/GetSIPOutboundTrunk": {sipAdmin: true},
|
||||
"livekit.SIP/ListSIPTrunk": {sipAdmin: true},
|
||||
"livekit.SIP/ListSIPInboundTrunk": {sipAdmin: true},
|
||||
"livekit.SIP/ListSIPOutboundTrunk": {sipAdmin: true},
|
||||
"livekit.SIP/DeleteSIPTrunk": {sipAdmin: true},
|
||||
"livekit.SIP/CreateSIPDispatchRule": {sipAdmin: true},
|
||||
"livekit.SIP/UpdateSIPDispatchRule": {sipAdmin: true},
|
||||
"livekit.SIP/ListSIPDispatchRule": {sipAdmin: true},
|
||||
"livekit.SIP/DeleteSIPDispatchRule": {sipAdmin: true},
|
||||
// Placing a call requires sip.call; transfer also requires room admin.
|
||||
"livekit.SIP/CreateSIPParticipant": {sipCall: true},
|
||||
"livekit.SIP/TransferSIPParticipant": {sipCall: true, roomAdmin: true},
|
||||
|
||||
// AgentDispatch — room admin scoped to the dispatch's room
|
||||
"livekit.AgentDispatchService/CreateDispatch": {roomAdmin: true},
|
||||
"livekit.AgentDispatchService/DeleteDispatch": {roomAdmin: true},
|
||||
"livekit.AgentDispatchService/ListDispatch": {roomAdmin: true},
|
||||
|
||||
// Connector (cloud) — initiating a call requires room create
|
||||
"livekit.Connector/DialWhatsAppCall": {roomCreate: true},
|
||||
"livekit.Connector/DisconnectWhatsAppCall": {roomCreate: true},
|
||||
"livekit.Connector/ConnectWhatsAppCall": {roomCreate: true},
|
||||
"livekit.Connector/AcceptWhatsAppCall": {roomCreate: true},
|
||||
"livekit.Connector/ConnectTwilioCall": {roomCreate: true},
|
||||
}
|
||||
|
||||
// authorize enforces the permissions a method requires. It returns the HTTP
|
||||
// status and Twirp error code to send (0, "" means authorized / not enforced).
|
||||
func (h *mockHandler) authorize(key string, r *http.Request, cfg *mockConfig, req proto.Message) (int, string) {
|
||||
if cfg.SkipAuth {
|
||||
return 0, ""
|
||||
}
|
||||
p, known := methodPerms[key]
|
||||
if !known {
|
||||
return 0, "" // unknown/future method: don't enforce
|
||||
}
|
||||
|
||||
grants, err := h.verifyToken(r.Header.Get("Authorization"))
|
||||
if err != nil {
|
||||
// Missing, malformed, or improperly-signed token, like the real server.
|
||||
return http.StatusUnauthorized, "unauthenticated"
|
||||
}
|
||||
if !p.satisfiedBy(grants, req) {
|
||||
return http.StatusForbidden, "permission_denied"
|
||||
}
|
||||
return 0, ""
|
||||
}
|
||||
|
||||
// verifyToken parses and verifies a "Bearer <jwt>" header using the protocol's
|
||||
// own auth helpers (the same path the real server uses), returning the grants.
|
||||
func (h *mockHandler) verifyToken(authorization string) (*auth.ClaimGrants, error) {
|
||||
token := strings.TrimSpace(strings.TrimPrefix(authorization, "Bearer "))
|
||||
v, err := auth.ParseAPIToken(token)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
_, grants, err := v.Verify(h.apiSecret)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return grants, nil
|
||||
}
|
||||
|
||||
func (p perm) satisfiedBy(g *auth.ClaimGrants, req proto.Message) bool {
|
||||
v, s := g.Video, g.SIP
|
||||
if p.roomCreate && (v == nil || !v.RoomCreate) {
|
||||
return false
|
||||
}
|
||||
if p.roomList && (v == nil || !v.RoomList) {
|
||||
return false
|
||||
}
|
||||
if p.roomRecord && (v == nil || !v.RoomRecord) {
|
||||
return false
|
||||
}
|
||||
if p.ingressAdmin && (v == nil || !v.IngressAdmin) {
|
||||
return false
|
||||
}
|
||||
if p.sipAdmin && (s == nil || !s.Admin) {
|
||||
return false
|
||||
}
|
||||
if p.sipCall && (s == nil || !s.Call) {
|
||||
return false
|
||||
}
|
||||
if p.roomAdmin || p.destRoom {
|
||||
if v == nil || !v.RoomAdmin {
|
||||
return false
|
||||
}
|
||||
if room := requestRoom(req); room != "" && v.Room != room {
|
||||
return false
|
||||
}
|
||||
}
|
||||
if p.destRoom {
|
||||
if dest := requestString(req, "destination_room"); dest != "" && v.DestinationRoom != dest {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// requestRoom reads the room name a request targets, trying the common "room"
|
||||
// and "room_name" fields.
|
||||
func requestRoom(req proto.Message) string {
|
||||
if v := requestString(req, "room"); v != "" {
|
||||
return v
|
||||
}
|
||||
return requestString(req, "room_name")
|
||||
}
|
||||
|
||||
func requestString(req proto.Message, field string) string {
|
||||
if req == nil {
|
||||
return ""
|
||||
}
|
||||
m := req.ProtoReflect()
|
||||
fd := m.Descriptor().Fields().ByName(protoreflect.Name(field))
|
||||
if fd == nil || fd.Kind() != protoreflect.StringKind || fd.IsList() {
|
||||
return ""
|
||||
}
|
||||
return m.Get(fd).String()
|
||||
}
|
||||
|
||||
func requestBool(req proto.Message, field string) bool {
|
||||
if req == nil {
|
||||
return false
|
||||
}
|
||||
m := req.ProtoReflect()
|
||||
fd := m.Descriptor().Fields().ByName(protoreflect.Name(field))
|
||||
if fd == nil || fd.Kind() != protoreflect.BoolKind || fd.IsList() {
|
||||
return false
|
||||
}
|
||||
return m.Get(fd).Bool()
|
||||
}
|
||||
@@ -0,0 +1,146 @@
|
||||
// Copyright 2026 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 main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
const (
|
||||
// headerMock carries the whole mock control config as a JSON object; see
|
||||
// mockConfig and the README. Sent by the SDK on API calls (and forwarded onto
|
||||
// the /settings/regions fetch and every failover retry).
|
||||
headerMock = "X-Lk-Mock"
|
||||
// headerRegion is set on responses to the index of the region that served it.
|
||||
headerRegion = "X-Lk-Mock-Region"
|
||||
)
|
||||
|
||||
// Deprecated: the individual X-Lk-Mock-* control headers predate the unified
|
||||
// X-Lk-Mock JSON header. They are still honored for existing clients; new
|
||||
// clients should send X-Lk-Mock instead. When X-Lk-Mock is present its fields
|
||||
// take precedence over any legacy header.
|
||||
const (
|
||||
legacyHeaderFailRegions = "X-Lk-Mock-Fail-Regions"
|
||||
legacyHeaderFailMode = "X-Lk-Mock-Fail-Mode"
|
||||
legacyHeaderFailStatus = "X-Lk-Mock-Fail-Status"
|
||||
legacyHeaderFailTwirpCode = "X-Lk-Mock-Fail-Twirp-Code"
|
||||
legacyHeaderDelayMs = "X-Lk-Mock-Delay-Ms"
|
||||
legacyHeaderRegionsStatus = "X-Lk-Mock-Regions-Status"
|
||||
legacyHeaderResponse = "X-Lk-Mock-Response"
|
||||
legacyHeaderSkipAuth = "X-Lk-Mock-Skip-Auth"
|
||||
)
|
||||
|
||||
// legacyDefaultDelayMs is the sleep used by the deprecated "delay" fail mode when
|
||||
// no X-Lk-Mock-Delay-Ms is given (long enough to trip client timeouts).
|
||||
const legacyDefaultDelayMs = 30_000
|
||||
|
||||
// mockConfig is the JSON value of the X-Lk-Mock request header. Every field is
|
||||
// optional; the zero value means "behave normally". A single object keeps the
|
||||
// control protocol simple — the SDK serializes one struct instead of juggling a
|
||||
// header per knob.
|
||||
type mockConfig struct {
|
||||
// FailRegions lists region indices that should fail this request. A listener
|
||||
// fails only if its own region index appears here, so one config can make the
|
||||
// primary fail while a fallback succeeds.
|
||||
FailRegions []int `json:"failRegions,omitempty"`
|
||||
// FailMode selects how a failing region fails: "status" (default) writes a
|
||||
// Twirp error; "drop" closes the connection to force a transport error.
|
||||
// ("delay" is a deprecated legacy mode; new clients use DelayMs instead.)
|
||||
FailMode string `json:"failMode,omitempty"`
|
||||
// FailStatus is the HTTP status for a "status"-mode failure (default 503).
|
||||
FailStatus int `json:"failStatus,omitempty"`
|
||||
// FailTwirpCode overrides the Twirp error code string in the failure body
|
||||
// (default derived from FailStatus).
|
||||
FailTwirpCode string `json:"failTwirpCode,omitempty"`
|
||||
// DelayMs delays the response by this many milliseconds before returning,
|
||||
// whether the region succeeds or fails. It overrides a method's natural
|
||||
// latency (see methodLatency): set it high for timeout tests, or to 0 to skip
|
||||
// a SIP method's built-in wait. Nil means "use the natural latency".
|
||||
DelayMs *int `json:"delayMs,omitempty"`
|
||||
// RegionsStatus overrides the HTTP status of GET /settings/regions (default 200).
|
||||
RegionsStatus int `json:"regionsStatus,omitempty"`
|
||||
// Response is the protojson of the response message for the called method; it
|
||||
// replaces the populated default, giving full control over the payload.
|
||||
Response json.RawMessage `json:"response,omitempty"`
|
||||
// SkipAuth disables permission enforcement for this request (for tests that
|
||||
// aren't about authz, e.g. failover tests with a placeholder token).
|
||||
SkipAuth bool `json:"skipAuth,omitempty"`
|
||||
// SIPStatus, when set on a SIP dial method (CreateSIPParticipant /
|
||||
// TransferSIPParticipant), fails the call with this SIP status. The Twirp
|
||||
// error code and metadata (sip_status_code, sip_status, error_details) are
|
||||
// derived from it exactly as the real server does, so the SDK sees an
|
||||
// identical error. Composes with DelayMs to simulate "ring, then fail".
|
||||
SIPStatus *sipStatusConfig `json:"sipStatus,omitempty"`
|
||||
|
||||
// legacyDelayMs is the sleep used by the deprecated "delay" fail mode. It is
|
||||
// populated only from the legacy X-Lk-Mock-Delay-Ms header, never from JSON.
|
||||
legacyDelayMs int
|
||||
}
|
||||
|
||||
// sipStatusConfig is a SIP response to inject; see mockConfig.SIPStatus.
|
||||
type sipStatusConfig struct {
|
||||
// Code is the SIP response code, e.g. 486 (Busy Here) or 603 (Decline).
|
||||
Code int `json:"code"`
|
||||
// Status is the SIP reason phrase; defaults to the code's canonical name.
|
||||
Status string `json:"status,omitempty"`
|
||||
}
|
||||
|
||||
// parseMockConfig builds the request's config. Deprecated individual X-Lk-Mock-*
|
||||
// headers form the base; the unified X-Lk-Mock JSON header (if present) is
|
||||
// overlaid on top, so its fields win per-field while absent fields keep the
|
||||
// legacy value.
|
||||
func parseMockConfig(r *http.Request) mockConfig {
|
||||
cfg := parseLegacyConfig(r)
|
||||
if v := r.Header.Get(headerMock); v != "" {
|
||||
// Unmarshal overwrites only the fields present in the JSON; the unexported
|
||||
// legacyDelayMs is untouched.
|
||||
_ = json.Unmarshal([]byte(v), &cfg)
|
||||
}
|
||||
return cfg
|
||||
}
|
||||
|
||||
// parseLegacyConfig reads the deprecated per-setting headers into a config.
|
||||
func parseLegacyConfig(r *http.Request) mockConfig {
|
||||
var cfg mockConfig
|
||||
for _, part := range strings.Split(r.Header.Get(legacyHeaderFailRegions), ",") {
|
||||
if idx, err := strconv.Atoi(strings.TrimSpace(part)); err == nil {
|
||||
cfg.FailRegions = append(cfg.FailRegions, idx)
|
||||
}
|
||||
}
|
||||
cfg.FailMode = r.Header.Get(legacyHeaderFailMode)
|
||||
cfg.FailStatus = parseStatus(r.Header.Get(legacyHeaderFailStatus))
|
||||
cfg.FailTwirpCode = r.Header.Get(legacyHeaderFailTwirpCode)
|
||||
cfg.RegionsStatus = parseStatus(r.Header.Get(legacyHeaderRegionsStatus))
|
||||
if resp := r.Header.Get(legacyHeaderResponse); resp != "" {
|
||||
cfg.Response = json.RawMessage(resp)
|
||||
}
|
||||
cfg.SkipAuth = strings.EqualFold(r.Header.Get(legacyHeaderSkipAuth), "true")
|
||||
cfg.legacyDelayMs = legacyDefaultDelayMs
|
||||
if ms, err := strconv.Atoi(r.Header.Get(legacyHeaderDelayMs)); err == nil && ms >= 0 {
|
||||
cfg.legacyDelayMs = ms
|
||||
}
|
||||
return cfg
|
||||
}
|
||||
|
||||
// parseStatus returns a valid HTTP status from s, or 0 if absent/invalid.
|
||||
func parseStatus(s string) int {
|
||||
if v, err := strconv.Atoi(strings.TrimSpace(s)); err == nil && v >= 100 && v <= 599 {
|
||||
return v
|
||||
}
|
||||
return 0
|
||||
}
|
||||
@@ -0,0 +1,314 @@
|
||||
// Copyright 2026 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 main
|
||||
|
||||
import (
|
||||
"io"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"google.golang.org/protobuf/proto"
|
||||
"google.golang.org/protobuf/reflect/protoreflect"
|
||||
"google.golang.org/protobuf/types/known/emptypb"
|
||||
|
||||
"github.com/livekit/protocol/livekit"
|
||||
"github.com/livekit/protocol/utils/protojson"
|
||||
"github.com/livekit/protocol/utils/xtwirp"
|
||||
)
|
||||
|
||||
// apiSpec captures the request and response message types for one Twirp method,
|
||||
// so the mock can decode the incoming request and build a typed response.
|
||||
type apiSpec struct {
|
||||
newReq func() proto.Message
|
||||
newResp func() proto.Message
|
||||
}
|
||||
|
||||
// ptrMsg constrains a pointer type that is also a proto.Message, letting reg
|
||||
// construct fresh request/response values generically.
|
||||
type ptrMsg[T any] interface {
|
||||
*T
|
||||
proto.Message
|
||||
}
|
||||
|
||||
// apiHandlers maps "<package>.<Service>/<Method>" to its message types. It
|
||||
// covers the full LiveKit API surface; see init below.
|
||||
var apiHandlers = map[string]apiSpec{}
|
||||
|
||||
func reg[ReqT, RespT any, Req ptrMsg[ReqT], Resp ptrMsg[RespT]](key string) {
|
||||
apiHandlers[key] = apiSpec{
|
||||
newReq: func() proto.Message { return Req(new(ReqT)) },
|
||||
newResp: func() proto.Message { return Resp(new(RespT)) },
|
||||
}
|
||||
}
|
||||
|
||||
func init() {
|
||||
// RoomService
|
||||
reg[livekit.CreateRoomRequest, livekit.Room]("livekit.RoomService/CreateRoom")
|
||||
reg[livekit.ListRoomsRequest, livekit.ListRoomsResponse]("livekit.RoomService/ListRooms")
|
||||
reg[livekit.DeleteRoomRequest, livekit.DeleteRoomResponse]("livekit.RoomService/DeleteRoom")
|
||||
reg[livekit.ListParticipantsRequest, livekit.ListParticipantsResponse]("livekit.RoomService/ListParticipants")
|
||||
reg[livekit.RoomParticipantIdentity, livekit.ParticipantInfo]("livekit.RoomService/GetParticipant")
|
||||
reg[livekit.RoomParticipantIdentity, livekit.RemoveParticipantResponse]("livekit.RoomService/RemoveParticipant")
|
||||
reg[livekit.MuteRoomTrackRequest, livekit.MuteRoomTrackResponse]("livekit.RoomService/MutePublishedTrack")
|
||||
reg[livekit.UpdateParticipantRequest, livekit.ParticipantInfo]("livekit.RoomService/UpdateParticipant")
|
||||
reg[livekit.UpdateSubscriptionsRequest, livekit.UpdateSubscriptionsResponse]("livekit.RoomService/UpdateSubscriptions")
|
||||
reg[livekit.SendDataRequest, livekit.SendDataResponse]("livekit.RoomService/SendData")
|
||||
reg[livekit.UpdateRoomMetadataRequest, livekit.Room]("livekit.RoomService/UpdateRoomMetadata")
|
||||
reg[livekit.ForwardParticipantRequest, livekit.ForwardParticipantResponse]("livekit.RoomService/ForwardParticipant")
|
||||
reg[livekit.MoveParticipantRequest, livekit.MoveParticipantResponse]("livekit.RoomService/MoveParticipant")
|
||||
reg[livekit.PerformRpcRequest, livekit.PerformRpcResponse]("livekit.RoomService/PerformRpc")
|
||||
|
||||
// Egress
|
||||
reg[livekit.StartEgressRequest, livekit.EgressInfo]("livekit.Egress/StartEgress")
|
||||
reg[livekit.UpdateLayoutRequest, livekit.EgressInfo]("livekit.Egress/UpdateLayout")
|
||||
reg[livekit.UpdateStreamRequest, livekit.EgressInfo]("livekit.Egress/UpdateStream")
|
||||
reg[livekit.ListEgressRequest, livekit.ListEgressResponse]("livekit.Egress/ListEgress")
|
||||
reg[livekit.StopEgressRequest, livekit.EgressInfo]("livekit.Egress/StopEgress")
|
||||
reg[livekit.RoomCompositeEgressRequest, livekit.EgressInfo]("livekit.Egress/StartRoomCompositeEgress")
|
||||
reg[livekit.WebEgressRequest, livekit.EgressInfo]("livekit.Egress/StartWebEgress")
|
||||
reg[livekit.ParticipantEgressRequest, livekit.EgressInfo]("livekit.Egress/StartParticipantEgress")
|
||||
reg[livekit.TrackCompositeEgressRequest, livekit.EgressInfo]("livekit.Egress/StartTrackCompositeEgress")
|
||||
reg[livekit.TrackEgressRequest, livekit.EgressInfo]("livekit.Egress/StartTrackEgress")
|
||||
|
||||
// Ingress
|
||||
reg[livekit.CreateIngressRequest, livekit.IngressInfo]("livekit.Ingress/CreateIngress")
|
||||
reg[livekit.UpdateIngressRequest, livekit.IngressInfo]("livekit.Ingress/UpdateIngress")
|
||||
reg[livekit.ListIngressRequest, livekit.ListIngressResponse]("livekit.Ingress/ListIngress")
|
||||
reg[livekit.DeleteIngressRequest, livekit.IngressInfo]("livekit.Ingress/DeleteIngress")
|
||||
|
||||
// SIP
|
||||
reg[livekit.ListSIPTrunkRequest, livekit.ListSIPTrunkResponse]("livekit.SIP/ListSIPTrunk")
|
||||
reg[livekit.CreateSIPInboundTrunkRequest, livekit.SIPInboundTrunkInfo]("livekit.SIP/CreateSIPInboundTrunk")
|
||||
reg[livekit.CreateSIPOutboundTrunkRequest, livekit.SIPOutboundTrunkInfo]("livekit.SIP/CreateSIPOutboundTrunk")
|
||||
reg[livekit.UpdateSIPInboundTrunkRequest, livekit.SIPInboundTrunkInfo]("livekit.SIP/UpdateSIPInboundTrunk")
|
||||
reg[livekit.UpdateSIPOutboundTrunkRequest, livekit.SIPOutboundTrunkInfo]("livekit.SIP/UpdateSIPOutboundTrunk")
|
||||
reg[livekit.GetSIPInboundTrunkRequest, livekit.GetSIPInboundTrunkResponse]("livekit.SIP/GetSIPInboundTrunk")
|
||||
reg[livekit.GetSIPOutboundTrunkRequest, livekit.GetSIPOutboundTrunkResponse]("livekit.SIP/GetSIPOutboundTrunk")
|
||||
reg[livekit.ListSIPInboundTrunkRequest, livekit.ListSIPInboundTrunkResponse]("livekit.SIP/ListSIPInboundTrunk")
|
||||
reg[livekit.ListSIPOutboundTrunkRequest, livekit.ListSIPOutboundTrunkResponse]("livekit.SIP/ListSIPOutboundTrunk")
|
||||
reg[livekit.DeleteSIPTrunkRequest, livekit.SIPTrunkInfo]("livekit.SIP/DeleteSIPTrunk")
|
||||
reg[livekit.CreateSIPDispatchRuleRequest, livekit.SIPDispatchRuleInfo]("livekit.SIP/CreateSIPDispatchRule")
|
||||
reg[livekit.UpdateSIPDispatchRuleRequest, livekit.SIPDispatchRuleInfo]("livekit.SIP/UpdateSIPDispatchRule")
|
||||
reg[livekit.ListSIPDispatchRuleRequest, livekit.ListSIPDispatchRuleResponse]("livekit.SIP/ListSIPDispatchRule")
|
||||
reg[livekit.DeleteSIPDispatchRuleRequest, livekit.SIPDispatchRuleInfo]("livekit.SIP/DeleteSIPDispatchRule")
|
||||
reg[livekit.CreateSIPParticipantRequest, livekit.SIPParticipantInfo]("livekit.SIP/CreateSIPParticipant")
|
||||
reg[livekit.TransferSIPParticipantRequest, emptypb.Empty]("livekit.SIP/TransferSIPParticipant")
|
||||
|
||||
// Connector
|
||||
reg[livekit.DialWhatsAppCallRequest, livekit.DialWhatsAppCallResponse]("livekit.Connector/DialWhatsAppCall")
|
||||
reg[livekit.DisconnectWhatsAppCallRequest, livekit.DisconnectWhatsAppCallResponse]("livekit.Connector/DisconnectWhatsAppCall")
|
||||
reg[livekit.ConnectWhatsAppCallRequest, livekit.ConnectWhatsAppCallResponse]("livekit.Connector/ConnectWhatsAppCall")
|
||||
reg[livekit.AcceptWhatsAppCallRequest, livekit.AcceptWhatsAppCallResponse]("livekit.Connector/AcceptWhatsAppCall")
|
||||
reg[livekit.ConnectTwilioCallRequest, livekit.ConnectTwilioCallResponse]("livekit.Connector/ConnectTwilioCall")
|
||||
}
|
||||
|
||||
// serveAPI handles a Twirp call end to end: decode the request, enforce the
|
||||
// method's required permissions (like the real server's auth middleware), apply
|
||||
// any region-failure injection, then serve a populated response.
|
||||
func (h *mockHandler) serveAPI(w http.ResponseWriter, r *http.Request) {
|
||||
json := strings.Contains(r.Header.Get("Content-Type"), "json")
|
||||
key := strings.TrimPrefix(r.URL.Path, h.twirpPrefix+"/")
|
||||
spec, known := apiHandlers[key]
|
||||
|
||||
// Decode the request up front — needed both to enforce room-scoped grants
|
||||
// and to build the echoed response.
|
||||
var req proto.Message
|
||||
if known {
|
||||
body, _ := io.ReadAll(r.Body)
|
||||
req = spec.newReq()
|
||||
if json {
|
||||
_ = protojson.Unmarshal(body, req)
|
||||
} else {
|
||||
_ = proto.Unmarshal(body, req)
|
||||
}
|
||||
}
|
||||
|
||||
cfg := parseMockConfig(r)
|
||||
|
||||
// Permission enforcement comes first, mirroring the real server.
|
||||
if status, code := h.authorize(key, r, &cfg, req); status != 0 {
|
||||
writeTwirpErrorCode(w, status, code, "mock: "+code)
|
||||
return
|
||||
}
|
||||
|
||||
// Delay before responding (success or failure). An explicit delayMs overrides
|
||||
// the method's natural latency — e.g. CreateSIPParticipant blocking until the
|
||||
// callee answers.
|
||||
delay := methodLatency(key, req)
|
||||
if cfg.DelayMs != nil {
|
||||
delay = time.Duration(*cfg.DelayMs) * time.Millisecond
|
||||
}
|
||||
if delay > 0 {
|
||||
time.Sleep(delay)
|
||||
}
|
||||
|
||||
// A SIP dial that fails carries a SIP status; the Twirp code and metadata are
|
||||
// derived from it exactly as the real server does.
|
||||
if cfg.SIPStatus != nil && isSIPDialMethod(key) {
|
||||
h.failSIP(w, &cfg)
|
||||
return
|
||||
}
|
||||
|
||||
if h.shouldFail(&cfg) {
|
||||
h.fail(w, &cfg)
|
||||
return
|
||||
}
|
||||
|
||||
h.writeAPIResponse(w, json, known, req, spec, &cfg)
|
||||
}
|
||||
|
||||
// methodLatency returns the realistic time a method blocks before responding, so
|
||||
// the mock approximates the real server's behavior. CreateSIPParticipant blocks
|
||||
// until the callee answers when wait_until_answered is set; TransferSIPParticipant
|
||||
// always blocks until the transfer (REFER) completes.
|
||||
func methodLatency(key string, req proto.Message) time.Duration {
|
||||
switch key {
|
||||
case "livekit.SIP/CreateSIPParticipant":
|
||||
if requestBool(req, "wait_until_answered") {
|
||||
return sipAnswerLatency
|
||||
}
|
||||
case "livekit.SIP/TransferSIPParticipant":
|
||||
return sipAnswerLatency
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// sipAnswerLatency is how long a SIP call takes to be answered/transferred in the
|
||||
// mock — long enough to exercise client-side timeouts around these calls.
|
||||
const sipAnswerLatency = 11 * time.Second
|
||||
|
||||
// isSIPDialMethod reports whether key places a call that can fail with a SIP status.
|
||||
func isSIPDialMethod(key string) bool {
|
||||
switch key {
|
||||
case "livekit.SIP/CreateSIPParticipant", "livekit.SIP/TransferSIPParticipant":
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// failSIP fails the request with the configured SIP status, mirroring the real
|
||||
// server: the status maps to a Twirp error code and attaches sip_status_code,
|
||||
// sip_status, and error_details metadata via xtwirp.
|
||||
func (h *mockHandler) failSIP(w http.ResponseWriter, cfg *mockConfig) {
|
||||
st := &livekit.SIPStatus{
|
||||
Code: livekit.SIPStatusCode(cfg.SIPStatus.Code),
|
||||
Status: cfg.SIPStatus.Status,
|
||||
}
|
||||
writeTwirpErr(w, xtwirp.ToError(st))
|
||||
}
|
||||
|
||||
// writeAPIResponse serves a populated, type-correct response for a known API
|
||||
// method. The response is the reflection-populated default unless the mock
|
||||
// config carries a `response` (protojson), which overrides it entirely. Content
|
||||
// type (protobuf vs JSON) mirrors the request.
|
||||
func (h *mockHandler) writeAPIResponse(w http.ResponseWriter, json, known bool, req proto.Message, spec apiSpec, cfg *mockConfig) {
|
||||
w.Header().Set(headerRegion, strconv.Itoa(h.regionIndex))
|
||||
|
||||
if !known {
|
||||
// Unknown/future method: an empty body still decodes to a valid default
|
||||
// message in every Twirp client.
|
||||
writeEmptySuccess(w, json)
|
||||
return
|
||||
}
|
||||
|
||||
resp := spec.newResp()
|
||||
if len(cfg.Response) > 0 {
|
||||
if err := protojson.Unmarshal(cfg.Response, resp); err != nil {
|
||||
// Malformed override: fall back to the populated default.
|
||||
resp = spec.newResp()
|
||||
populateMessage(resp.ProtoReflect(), req.ProtoReflect(), 1)
|
||||
}
|
||||
} else {
|
||||
populateMessage(resp.ProtoReflect(), req.ProtoReflect(), 1)
|
||||
}
|
||||
|
||||
if json {
|
||||
out, _ := protojson.Marshal(resp)
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_, _ = w.Write(out)
|
||||
} else {
|
||||
out, _ := proto.Marshal(resp)
|
||||
w.Header().Set("Content-Type", "application/protobuf")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_, _ = w.Write(out)
|
||||
}
|
||||
}
|
||||
|
||||
func writeEmptySuccess(w http.ResponseWriter, json bool) {
|
||||
if json {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_, _ = w.Write([]byte("{}"))
|
||||
} else {
|
||||
w.Header().Set("Content-Type", "application/protobuf")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}
|
||||
}
|
||||
|
||||
// populateMessage fills a response message with plausible values: it echoes
|
||||
// scalar fields that share a name with the request, assigns placeholder values
|
||||
// to id/sid fields, and adds one element to repeated-message (list) fields so
|
||||
// list endpoints return non-empty results. depth bounds list-element nesting.
|
||||
func populateMessage(m protoreflect.Message, req protoreflect.Message, depth int) {
|
||||
fields := m.Descriptor().Fields()
|
||||
for i := 0; i < fields.Len(); i++ {
|
||||
fd := fields.Get(i)
|
||||
|
||||
// Echo a same-named scalar field from the request (e.g. name, metadata,
|
||||
// identity, room, timeouts).
|
||||
if req != nil && fd.Cardinality() != protoreflect.Repeated && isScalarKind(fd.Kind()) {
|
||||
if rf := req.Descriptor().Fields().ByName(fd.Name()); rf != nil &&
|
||||
rf.Kind() == fd.Kind() && rf.Cardinality() != protoreflect.Repeated && req.Has(rf) {
|
||||
m.Set(fd, req.Get(rf))
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
// Give id/sid-like string fields a deterministic placeholder.
|
||||
if fd.Kind() == protoreflect.StringKind && fd.Cardinality() != protoreflect.Repeated && !m.Has(fd) {
|
||||
n := string(fd.Name())
|
||||
if n == "id" || n == "sid" || strings.HasSuffix(n, "_id") || strings.HasSuffix(n, "_sid") {
|
||||
m.Set(fd, protoreflect.ValueOfString("MOCK_"+strings.ToUpper(n)))
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
// Populate list endpoints with a single element so clients see results.
|
||||
if depth > 0 && fd.IsList() && fd.Kind() == protoreflect.MessageKind {
|
||||
list := m.Mutable(fd).List()
|
||||
elem := list.NewElement()
|
||||
populateMessage(elem.Message(), nil, depth-1)
|
||||
list.Append(elem)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func isScalarKind(k protoreflect.Kind) bool {
|
||||
switch k {
|
||||
case protoreflect.BoolKind,
|
||||
protoreflect.Int32Kind, protoreflect.Sint32Kind, protoreflect.Uint32Kind,
|
||||
protoreflect.Int64Kind, protoreflect.Sint64Kind, protoreflect.Uint64Kind,
|
||||
protoreflect.Sfixed32Kind, protoreflect.Fixed32Kind,
|
||||
protoreflect.Sfixed64Kind, protoreflect.Fixed64Kind,
|
||||
protoreflect.FloatKind, protoreflect.DoubleKind,
|
||||
protoreflect.StringKind, protoreflect.BytesKind:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,241 @@
|
||||
// Copyright 2026 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.
|
||||
|
||||
// Command test-server is a programmable mock of the LiveKit server HTTP API,
|
||||
// used by the server SDKs to test client behavior. See cmd/test-server/README.md.
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/twitchtv/twirp"
|
||||
|
||||
"github.com/livekit/protocol/livekit"
|
||||
"github.com/livekit/protocol/utils/protojson"
|
||||
)
|
||||
|
||||
func main() {
|
||||
portsFlag := flagValue("--ports", "LK_TEST_SERVER_PORTS", "9999,10000,10001,10002")
|
||||
advertiseHost := flagValue("--advertise-host", "LK_TEST_SERVER_ADVERTISE_HOST", "http://127.0.0.1")
|
||||
bindAddr := flagValue("--bind", "LK_TEST_SERVER_BIND", "0.0.0.0")
|
||||
twirpPrefix := flagValue("--twirp-prefix", "LK_TEST_SERVER_TWIRP_PREFIX", "/twirp")
|
||||
// API secret used to verify request tokens for permission enforcement.
|
||||
// Defaults to the `livekit-server --dev` secret.
|
||||
apiSecret := flagValue("--api-secret", "LK_TEST_SERVER_API_SECRET", "secret")
|
||||
|
||||
ports, err := parsePorts(portsFlag)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "invalid --ports: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
advertiseHost = strings.TrimRight(advertiseHost, "/")
|
||||
|
||||
regions := &livekit.RegionSettings{}
|
||||
for i, p := range ports {
|
||||
regions.Regions = append(regions.Regions, &livekit.RegionInfo{
|
||||
Region: fmt.Sprintf("region-%d", i),
|
||||
Url: fmt.Sprintf("%s:%d", advertiseHost, p),
|
||||
Distance: int64(i),
|
||||
})
|
||||
}
|
||||
|
||||
errCh := make(chan error, len(ports))
|
||||
for i, p := range ports {
|
||||
srv := &http.Server{
|
||||
Addr: fmt.Sprintf("%s:%d", bindAddr, p),
|
||||
Handler: &mockHandler{regionIndex: i, regions: regions, twirpPrefix: twirpPrefix, apiSecret: apiSecret},
|
||||
}
|
||||
go func() { errCh <- srv.ListenAndServe() }()
|
||||
fmt.Printf("test-server: region-%d listening on %s:%d (advertised as %s:%d)\n", i, bindAddr, p, advertiseHost, p)
|
||||
}
|
||||
|
||||
sigCh := make(chan os.Signal, 1)
|
||||
signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
|
||||
select {
|
||||
case err := <-errCh:
|
||||
fmt.Fprintf(os.Stderr, "listener failed: %v\n", err)
|
||||
os.Exit(1)
|
||||
case <-sigCh:
|
||||
fmt.Println("test-server: shutting down")
|
||||
}
|
||||
}
|
||||
|
||||
type mockHandler struct {
|
||||
regionIndex int
|
||||
regions *livekit.RegionSettings
|
||||
twirpPrefix string
|
||||
apiSecret string
|
||||
}
|
||||
|
||||
func (h *mockHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
switch {
|
||||
case r.URL.Path == "/settings/regions":
|
||||
h.handleRegions(w, r)
|
||||
case strings.HasPrefix(r.URL.Path, h.twirpPrefix+"/"):
|
||||
h.handleTwirp(w, r)
|
||||
case r.URL.Path == "/" || r.URL.Path == "/_test/health":
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_, _ = w.Write([]byte("ok"))
|
||||
default:
|
||||
http.NotFound(w, r)
|
||||
}
|
||||
}
|
||||
|
||||
func (h *mockHandler) handleRegions(w http.ResponseWriter, r *http.Request) {
|
||||
cfg := parseMockConfig(r)
|
||||
if cfg.RegionsStatus != 0 && cfg.RegionsStatus != http.StatusOK {
|
||||
w.WriteHeader(cfg.RegionsStatus)
|
||||
return
|
||||
}
|
||||
body, err := protojson.Marshal(h.regions)
|
||||
if err != nil {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.Header().Set("Cache-Control", "max-age=0")
|
||||
w.Header().Set(headerRegion, strconv.Itoa(h.regionIndex))
|
||||
_, _ = w.Write(body)
|
||||
}
|
||||
|
||||
func (h *mockHandler) handleTwirp(w http.ResponseWriter, r *http.Request) {
|
||||
h.serveAPI(w, r)
|
||||
}
|
||||
|
||||
func (h *mockHandler) shouldFail(cfg *mockConfig) bool {
|
||||
return slices.Contains(cfg.FailRegions, h.regionIndex)
|
||||
}
|
||||
|
||||
func (h *mockHandler) fail(w http.ResponseWriter, cfg *mockConfig) {
|
||||
switch strings.ToLower(cfg.FailMode) {
|
||||
case "drop":
|
||||
if hj, ok := w.(http.Hijacker); ok {
|
||||
if conn, _, err := hj.Hijack(); err == nil {
|
||||
_ = conn.Close()
|
||||
return
|
||||
}
|
||||
}
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
return
|
||||
case "delay":
|
||||
// Deprecated legacy mode: sleep, then status-fail. New clients should set
|
||||
// DelayMs (which delays every response) instead.
|
||||
time.Sleep(time.Duration(cfg.legacyDelayMs) * time.Millisecond)
|
||||
}
|
||||
status := cfg.FailStatus
|
||||
if status < 100 || status > 599 {
|
||||
status = http.StatusServiceUnavailable
|
||||
}
|
||||
writeTwirpError(w, cfg, status)
|
||||
}
|
||||
|
||||
func writeTwirpError(w http.ResponseWriter, cfg *mockConfig, status int) {
|
||||
code := cfg.FailTwirpCode
|
||||
if code == "" {
|
||||
code = twirpCodeForStatus(status)
|
||||
}
|
||||
writeTwirpErrorCode(w, status, code, fmt.Sprintf("mock failure (status %d)", status))
|
||||
}
|
||||
|
||||
// writeTwirpErrorCode writes a Twirp JSON error with an explicit code and message.
|
||||
func writeTwirpErrorCode(w http.ResponseWriter, status int, code, msg string) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.Header().Set(headerRegion, "")
|
||||
w.WriteHeader(status)
|
||||
_, _ = fmt.Fprintf(w, `{"code":%q,"msg":%q}`, code, msg)
|
||||
}
|
||||
|
||||
// writeTwirpErr writes a full Twirp JSON error — code, message, and metadata —
|
||||
// using the HTTP status Twirp derives from the error code.
|
||||
func writeTwirpErr(w http.ResponseWriter, terr twirp.Error) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.Header().Set(headerRegion, "")
|
||||
w.WriteHeader(twirp.ServerHTTPStatusFromErrorCode(terr.Code()))
|
||||
_ = json.NewEncoder(w).Encode(struct {
|
||||
Code string `json:"code"`
|
||||
Msg string `json:"msg"`
|
||||
Meta map[string]string `json:"meta,omitempty"`
|
||||
}{
|
||||
Code: string(terr.Code()),
|
||||
Msg: terr.Msg(),
|
||||
Meta: terr.MetaMap(),
|
||||
})
|
||||
}
|
||||
|
||||
func twirpCodeForStatus(status int) string {
|
||||
switch {
|
||||
case status == http.StatusBadRequest:
|
||||
return "invalid_argument"
|
||||
case status == http.StatusUnauthorized:
|
||||
return "unauthenticated"
|
||||
case status == http.StatusForbidden:
|
||||
return "permission_denied"
|
||||
case status == http.StatusNotFound:
|
||||
return "not_found"
|
||||
case status == http.StatusTooManyRequests:
|
||||
return "resource_exhausted"
|
||||
case status >= 500:
|
||||
return "unavailable"
|
||||
default:
|
||||
return "internal"
|
||||
}
|
||||
}
|
||||
|
||||
func parsePorts(s string) ([]int, error) {
|
||||
var ports []int
|
||||
for _, part := range strings.Split(s, ",") {
|
||||
part = strings.TrimSpace(part)
|
||||
if part == "" {
|
||||
continue
|
||||
}
|
||||
v, err := strconv.Atoi(part)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%q is not a port number", part)
|
||||
}
|
||||
ports = append(ports, v)
|
||||
}
|
||||
if len(ports) == 0 {
|
||||
return nil, errors.New("at least one port is required")
|
||||
}
|
||||
return ports, nil
|
||||
}
|
||||
|
||||
// flagValue resolves a setting from a --flag, then an environment variable, then a default.
|
||||
func flagValue(flag, env, def string) string {
|
||||
prefix := flag + "="
|
||||
for i, arg := range os.Args[1:] {
|
||||
if arg == flag {
|
||||
if i+2 <= len(os.Args[1:]) {
|
||||
return os.Args[1:][i+1]
|
||||
}
|
||||
}
|
||||
if strings.HasPrefix(arg, prefix) {
|
||||
return strings.TrimPrefix(arg, prefix)
|
||||
}
|
||||
}
|
||||
if v := os.Getenv(env); v != "" {
|
||||
return v
|
||||
}
|
||||
return def
|
||||
}
|
||||
@@ -21,7 +21,7 @@ require (
|
||||
github.com/jxskiss/base62 v1.1.0
|
||||
github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731
|
||||
github.com/livekit/mediatransportutil v0.0.0-20260608063931-a3417d38cda0
|
||||
github.com/livekit/protocol v1.48.1-0.20260623210753-2e1bfd81dd63
|
||||
github.com/livekit/protocol v1.48.2
|
||||
github.com/livekit/psrpc v0.7.2
|
||||
github.com/mackerelio/go-osstat v0.2.7
|
||||
github.com/magefile/mage v1.17.2
|
||||
@@ -30,17 +30,17 @@ require (
|
||||
github.com/moby/moby/client v0.4.1
|
||||
github.com/olekukonko/tablewriter v1.1.4
|
||||
github.com/ory/dockertest/v4 v4.0.0
|
||||
github.com/pion/datachannel v1.6.0
|
||||
github.com/pion/datachannel v1.6.2
|
||||
github.com/pion/dtls/v3 v3.1.4
|
||||
github.com/pion/ice/v4 v4.2.7
|
||||
github.com/pion/interceptor v0.1.45
|
||||
github.com/pion/rtcp v1.2.16
|
||||
github.com/pion/rtp v1.10.2
|
||||
github.com/pion/sctp v1.9.5
|
||||
github.com/pion/sctp v1.10.3
|
||||
github.com/pion/sdp/v3 v3.0.19
|
||||
github.com/pion/transport/v4 v4.0.2
|
||||
github.com/pion/turn/v5 v5.0.10
|
||||
github.com/pion/webrtc/v4 v4.2.11
|
||||
github.com/pion/webrtc/v4 v4.2.16
|
||||
github.com/pkg/errors v0.9.1
|
||||
github.com/prometheus/client_golang v1.23.2
|
||||
github.com/redis/go-redis/v9 v9.21.0
|
||||
@@ -140,7 +140,7 @@ require (
|
||||
github.com/prometheus/client_model v0.6.2 // indirect
|
||||
github.com/prometheus/common v0.69.0 // indirect
|
||||
github.com/prometheus/procfs v0.20.1 // indirect
|
||||
github.com/urfave/cli/v3 v3.9.0
|
||||
github.com/urfave/cli/v3 v3.10.0
|
||||
github.com/wlynxg/anet v0.0.5 // indirect
|
||||
github.com/zeebo/xxh3 v1.1.0 // indirect
|
||||
go.uber.org/zap/exp v0.3.0 // indirect
|
||||
|
||||
@@ -160,8 +160,8 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT
|
||||
github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ=
|
||||
github.com/livekit/mediatransportutil v0.0.0-20260608063931-a3417d38cda0 h1:XHNNzebIKZRkLimla/hFGrAIX5EMWHctrgt3hLw7s+I=
|
||||
github.com/livekit/mediatransportutil v0.0.0-20260608063931-a3417d38cda0/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU=
|
||||
github.com/livekit/protocol v1.48.1-0.20260623210753-2e1bfd81dd63 h1:Rj9/54oztXeioAwUkukXBwcPE/GxL97WylFd1m7V2pQ=
|
||||
github.com/livekit/protocol v1.48.1-0.20260623210753-2e1bfd81dd63/go.mod h1:jO+y05AU9Ec4JswDyuzKCZ4bhziOS0CzMqgnbj60Dzs=
|
||||
github.com/livekit/protocol v1.48.2 h1:1Jv1Eckf2jMN7SgJm7fQkSRQoMkGpmuN63mFeh3gd1U=
|
||||
github.com/livekit/protocol v1.48.2/go.mod h1:jO+y05AU9Ec4JswDyuzKCZ4bhziOS0CzMqgnbj60Dzs=
|
||||
github.com/livekit/psrpc v0.7.2 h1:6oZ+NODJ2pLyaT6VqDq1F4Qc/3TpDUSpyphj/P9MhQc=
|
||||
github.com/livekit/psrpc v0.7.2/go.mod h1:rAI+m2+/cb4x9RXhLRtUx5ZwdfjjXOl4zi46IjEetaw=
|
||||
github.com/mackerelio/go-osstat v0.2.7 h1:TCavZi10wF49bT6iQZ9eT2keGZQpC69MTDfdJej5e94=
|
||||
@@ -229,8 +229,8 @@ github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJw
|
||||
github.com/opencontainers/image-spec v1.1.1/go.mod h1:qpqAh3Dmcf36wStyyWU+kCeDgrGnAve2nCC8+7h8Q0M=
|
||||
github.com/ory/dockertest/v4 v4.0.0 h1:i19aFsO/VXE0VrMk4ifnKW4G/KIJ93PCjLOslxXoPME=
|
||||
github.com/ory/dockertest/v4 v4.0.0/go.mod h1:b5Ofu8VIxWNhXFvQcLu17pRNQdoUBKtXBW74G4Ygzx8=
|
||||
github.com/pion/datachannel v1.6.0 h1:XecBlj+cvsxhAMZWFfFcPyUaDZtd7IJvrXqlXD/53i0=
|
||||
github.com/pion/datachannel v1.6.0/go.mod h1:ur+wzYF8mWdC+Mkis5Thosk+u/VOL287apDNEbFpsIk=
|
||||
github.com/pion/datachannel v1.6.2 h1:7EXQ8TH3vTouBUdRWYbcX2edSx9Yj6k5zl5P+qyxEPc=
|
||||
github.com/pion/datachannel v1.6.2/go.mod h1:pzbdAZvyGtXbcHM1hBbsFaOTf40lZizU/dNlvVOak6E=
|
||||
github.com/pion/dtls/v3 v3.1.4 h1:QhvtMflMfu9Kf0RcDC5BJBle4caPskByrKQR6uuYqpY=
|
||||
github.com/pion/dtls/v3 v3.1.4/go.mod h1:cr/qotLISUw/9C1m83ZPNZtj9WnXkYLpfCptPqbkInc=
|
||||
github.com/pion/ice/v4 v4.2.7 h1:zDEbC6MiEdhQpF8TxBOTws+NU6ZgGpveHrQq4Lc1kao=
|
||||
@@ -247,8 +247,8 @@ github.com/pion/rtcp v1.2.16 h1:fk1B1dNW4hsI78XUCljZJlC4kZOPk67mNRuQ0fcEkSo=
|
||||
github.com/pion/rtcp v1.2.16/go.mod h1:/as7VKfYbs5NIb4h6muQ35kQF/J0ZVNz2Z3xKoCBYOo=
|
||||
github.com/pion/rtp v1.10.2 h1:l+f6tTDcAH6xwepaAoW791ddhuYsJlqRATOzirO04Mo=
|
||||
github.com/pion/rtp v1.10.2/go.mod h1:Au8fc6cEByy8RLTwKTQTEeQqDB/SJDxwL4mZuxYA5Pk=
|
||||
github.com/pion/sctp v1.9.5 h1:QoSFB/drmAsmSeSFNQNI3xx010nW4HsycCZckRVWWag=
|
||||
github.com/pion/sctp v1.9.5/go.mod h1:N20Dq6LY+JvJDAh9VVh1JELngb2rQ8dPgds5yBWiPgw=
|
||||
github.com/pion/sctp v1.10.3 h1:1gBtLMA9lmwNuJkZSZJCdD5/Hz4yJs+7dAqi6ZY97QI=
|
||||
github.com/pion/sctp v1.10.3/go.mod h1:7KFmTwLcoYgJs/Z+99nJvsWL0qDpuyloSI0RbAqlrz0=
|
||||
github.com/pion/sdp/v3 v3.0.19 h1:1VMKs3gIkTQV5M3hNKfTAPrDXSNrYtOlmOD8+mSZUGQ=
|
||||
github.com/pion/sdp/v3 v3.0.19/go.mod h1:dE5WOSlzXrtiE/iuZqe9n+AcEbOjtAd3k5m5NtlV/qU=
|
||||
github.com/pion/srtp/v3 v3.0.12 h1:U7V17bckl7sI4mb3sepiojByDuBY0wNCqQE+6IlQBbc=
|
||||
@@ -259,12 +259,10 @@ github.com/pion/transport/v3 v3.1.1 h1:Tr684+fnnKlhPceU+ICdrw6KKkTms+5qHMgw6bIkY
|
||||
github.com/pion/transport/v3 v3.1.1/go.mod h1:+c2eewC5WJQHiAA46fkMMzoYZSuGzA/7E2FPrOYHctQ=
|
||||
github.com/pion/transport/v4 v4.0.2 h1:ifYlPqNwsy6aKQ9y8yzxXlHae5431ZrH2avkD/Rn6Tk=
|
||||
github.com/pion/transport/v4 v4.0.2/go.mod h1:06hFI+jCFcok2X2MekVufNZ/uzNZXivGBPfviSVcjgM=
|
||||
github.com/pion/turn/v4 v4.1.4 h1:EU11yMXKIsK43FhcUnjLlrhE4nboHZq+TXBIi3QpcxQ=
|
||||
github.com/pion/turn/v4 v4.1.4/go.mod h1:ES1DXVFKnOhuDkqn9hn5VJlSWmZPaRJLyBXoOeO/BmQ=
|
||||
github.com/pion/turn/v5 v5.0.10 h1:mOMZjudflXpte5OsCnXztpUKwNXcpXIAzMBnq9TXOSQ=
|
||||
github.com/pion/turn/v5 v5.0.10/go.mod h1:u3XjBqy2Z4+NhCUpDoOSsNuQDrPLvKStlCGWk6sTQ1E=
|
||||
github.com/pion/webrtc/v4 v4.2.11 h1:QUX1QZKlNIn4O7U5JxLPGP0sV5RTncZkzu9SPR3jVNU=
|
||||
github.com/pion/webrtc/v4 v4.2.11/go.mod h1:s/rAiyy77GyRFrZMx+Ls6aua26dIBPudH8/ZHYbIRWY=
|
||||
github.com/pion/webrtc/v4 v4.2.16 h1:oK1GAg0TWJtZWYB8J/BgTgGWPoV2148gQWocH12vr3Q=
|
||||
github.com/pion/webrtc/v4 v4.2.16/go.mod h1:y4HjLAkX90LH+C/qPqGOUgz8RA8CbDj3Iar3d+2hdKQ=
|
||||
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
|
||||
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
@@ -305,8 +303,8 @@ github.com/twitchtv/twirp v8.1.3+incompatible h1:+F4TdErPgSUbMZMwp13Q/KgDVuI7HJX
|
||||
github.com/twitchtv/twirp v8.1.3+incompatible/go.mod h1:RRJoFSAmTEh2weEqWtpPE3vFK5YBhA6bqp2l1kfCC5A=
|
||||
github.com/ua-parser/uap-go v0.0.0-20260529044130-17c35e68e58c h1:XbG4n3OWA1PcRTpbBA22E2ChPLvJCuwYRXO12tIyVL0=
|
||||
github.com/ua-parser/uap-go v0.0.0-20260529044130-17c35e68e58c/go.mod h1:gwANdYmo9R8LLwGnyDFWK2PMsaXXX2HhAvCnb/UhZsM=
|
||||
github.com/urfave/cli/v3 v3.9.0 h1:AV9lIiPv3ukYnxunaCUsHnEozptYmDN2F0+yWqLMn/c=
|
||||
github.com/urfave/cli/v3 v3.9.0/go.mod h1:ysVLtOEmg2tOy6PknnYVhDoouyC/6N42TMeoMzskhso=
|
||||
github.com/urfave/cli/v3 v3.10.0 h1:0aU8yOObVDMkM13Cj4G+zb4P0PdeJMec65f81Ak1ioM=
|
||||
github.com/urfave/cli/v3 v3.10.0/go.mod h1:ysVLtOEmg2tOy6PknnYVhDoouyC/6N42TMeoMzskhso=
|
||||
github.com/urfave/negroni/v3 v3.1.1 h1:6MS4nG9Jk/UuCACaUlNXCbiKa0ywF9LXz5dGu09v8hw=
|
||||
github.com/urfave/negroni/v3 v3.1.1/go.mod h1:jWvnX03kcSjDBl/ShB0iHvx5uOs7mAzZXW+JvJ5XYAs=
|
||||
github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU=
|
||||
|
||||
@@ -177,6 +177,11 @@ func TestAll() error {
|
||||
return mageutil.Run(context.Background(), "go test ./... -count=1 -timeout=4m -v")
|
||||
}
|
||||
|
||||
// runs the SDK test server (cmd/test-server) in the foreground
|
||||
func TestServer() error {
|
||||
return mageutil.Run(context.Background(), "go run ./cmd/test-server")
|
||||
}
|
||||
|
||||
// runs golangci-lint
|
||||
func Lint() error {
|
||||
if _, err := exec.LookPath("golangci-lint"); err != nil {
|
||||
|
||||
+30
-7
@@ -298,6 +298,8 @@ type LimitConfig struct {
|
||||
|
||||
MaxDataBlobKeyLength int `yaml:"max_data_blob_key_length,omitempty"`
|
||||
MaxDataBlobSize uint32 `yaml:"max_data_blobs_size,omitempty"`
|
||||
|
||||
MaxDataTrackCustomEncodingLength int `yaml:"max_data_track_custom_encoding_length,omitempty"`
|
||||
}
|
||||
|
||||
func (l LimitConfig) CheckRoomNameLength(name string) bool {
|
||||
@@ -332,6 +334,26 @@ func (l LimitConfig) CheckDataBlobKeyLength(key string) bool {
|
||||
return l.MaxDataBlobKeyLength == 0 || len(key) <= l.MaxDataBlobKeyLength
|
||||
}
|
||||
|
||||
func (l LimitConfig) CheckDataTrackCustomEncodingLength(identifier string) bool {
|
||||
return l.MaxDataTrackCustomEncodingLength == 0 || len(identifier) <= l.MaxDataTrackCustomEncodingLength
|
||||
}
|
||||
|
||||
func (l LimitConfig) CheckDataTrackFrameEncoding(encoding *livekit.DataTrackFrameEncoding) bool {
|
||||
custom, ok := encoding.GetValue().(*livekit.DataTrackFrameEncoding_Custom)
|
||||
if !ok {
|
||||
return true
|
||||
}
|
||||
return len(custom.Custom) != 0 && l.CheckDataTrackCustomEncodingLength(custom.Custom)
|
||||
}
|
||||
|
||||
func (l LimitConfig) CheckDataTrackSchemaID(schema *livekit.DataTrackSchemaId) bool {
|
||||
custom, ok := schema.GetEncoding().GetValue().(*livekit.DataTrackSchemaEncoding_Custom)
|
||||
if !ok {
|
||||
return true
|
||||
}
|
||||
return len(custom.Custom) != 0 && l.CheckDataTrackCustomEncodingLength(custom.Custom)
|
||||
}
|
||||
|
||||
func (l LimitConfig) CheckDataBlobsSize(dataBlobs []*livekit.DataBlob) bool {
|
||||
if l.MaxDataBlobSize == 0 {
|
||||
return true
|
||||
@@ -475,13 +497,14 @@ var DefaultConfig = Config{
|
||||
UpdateBatchTargetSize: 128 * 1024,
|
||||
},
|
||||
Limit: LimitConfig{
|
||||
MaxMetadataSize: 512 * 1024,
|
||||
MaxAttributesSize: 64 * 1024,
|
||||
MaxRoomNameLength: 256,
|
||||
MaxParticipantIdentityLength: 256,
|
||||
MaxParticipantNameLength: 256,
|
||||
MaxDataBlobKeyLength: 256,
|
||||
MaxDataBlobSize: 64000,
|
||||
MaxMetadataSize: 512 * 1024,
|
||||
MaxAttributesSize: 64 * 1024,
|
||||
MaxRoomNameLength: 256,
|
||||
MaxParticipantIdentityLength: 256,
|
||||
MaxParticipantNameLength: 256,
|
||||
MaxDataBlobKeyLength: 256,
|
||||
MaxDataBlobSize: 64000,
|
||||
MaxDataTrackCustomEncodingLength: 32,
|
||||
},
|
||||
Logging: LoggingConfig{
|
||||
PionLevel: "error",
|
||||
|
||||
+11
-6
@@ -227,6 +227,7 @@ type ParticipantParams struct {
|
||||
DisableTransceiverReuseForE2EE bool
|
||||
EnableParticipantDataBlob bool
|
||||
EnableStartAtDesiredQuality bool
|
||||
MigrationWaitDuration time.Duration
|
||||
}
|
||||
|
||||
type ParticipantImpl struct {
|
||||
@@ -1426,19 +1427,23 @@ func (p *ParticipantImpl) IsReconnect() bool {
|
||||
}
|
||||
|
||||
func (p *ParticipantImpl) maybeRecordRTCanceled(closeReason types.ParticipantCloseReason) {
|
||||
if p.State() >= livekit.ParticipantInfo_ACTIVE {
|
||||
if p.HasConnected() {
|
||||
return
|
||||
}
|
||||
|
||||
if closeReason == types.ParticipantCloseReasonClientRequestLeave ||
|
||||
if p.IsConnectionCanceled(closeReason) {
|
||||
prometheus.IncrementParticipantRtcCanceled(1)
|
||||
}
|
||||
}
|
||||
|
||||
func (p *ParticipantImpl) IsConnectionCanceled(closeReason types.ParticipantCloseReason) bool {
|
||||
return closeReason == types.ParticipantCloseReasonClientRequestLeave ||
|
||||
closeReason == types.ParticipantCloseReasonDuplicateIdentity ||
|
||||
closeReason == types.ParticipantCloseReasonRoomClosed ||
|
||||
closeReason == types.ParticipantCloseReasonMigrationRequested ||
|
||||
closeReason == types.ParticipantCloseReasonMigrationComplete ||
|
||||
// client closing signal connection too quickly, there is a time check to handle clients timing out and leaving without sending a leave message
|
||||
(time.Since(p.params.SessionStartTime) < 3*time.Second && closeReason == types.ParticipantCloseReasonSignalSourceClose) {
|
||||
prometheus.IncrementParticipantRtcCanceled(1)
|
||||
}
|
||||
(time.Since(p.params.SessionStartTime) < 3*time.Second && closeReason == types.ParticipantCloseReasonSignalSourceClose)
|
||||
}
|
||||
|
||||
func (p *ParticipantImpl) Close(sendLeave bool, reason types.ParticipantCloseReason, isExpectedToResume bool) error {
|
||||
@@ -1572,7 +1577,7 @@ func (p *ParticipantImpl) setupMigrationTimerLocked() {
|
||||
// to try and succeed. If not, close the subscriber peer connection
|
||||
// and help the remote side to narrow down its ICE candidate pool.
|
||||
//
|
||||
p.migrationTimer = time.AfterFunc(migrationWaitDuration, func() {
|
||||
p.migrationTimer = time.AfterFunc(max(p.params.MigrationWaitDuration, migrationWaitDuration), func() {
|
||||
p.clearMigrationTimer()
|
||||
|
||||
if p.IsClosed() || p.IsDisconnected() {
|
||||
|
||||
@@ -41,6 +41,16 @@ func (p *ParticipantImpl) HandleStoreDataBlobRequest(req *livekit.StoreDataBlobR
|
||||
return
|
||||
}
|
||||
|
||||
if !p.params.LimitConfig.CheckDataTrackSchemaID(req.Blob.Key.GetSchemaId()) {
|
||||
p.pubLogger.Warnw("data blob key schema id is invalid", nil, "req", logger.Proto(req))
|
||||
p.sendRequestResponse(&livekit.RequestResponse{
|
||||
RequestId: req.RequestId,
|
||||
Reason: livekit.RequestResponse_INVALID_REQUEST,
|
||||
Message: "encoding identifier is empty or exceeds the maximum length",
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
if len(req.Blob.Contents) == 0 {
|
||||
p.sendRequestResponse(&livekit.RequestResponse{
|
||||
RequestId: req.RequestId,
|
||||
|
||||
@@ -60,6 +60,19 @@ func (p *ParticipantImpl) HandlePublishDataTrackRequest(req *livekit.PublishData
|
||||
return
|
||||
}
|
||||
|
||||
if !p.params.LimitConfig.CheckDataTrackFrameEncoding(req.FrameEncoding) ||
|
||||
!p.params.LimitConfig.CheckDataTrackSchemaID(req.Schema) {
|
||||
p.pubLogger.Warnw("invalid encoding identifier", nil, "req", logger.Proto(req))
|
||||
p.sendRequestResponse(&livekit.RequestResponse{
|
||||
Reason: livekit.RequestResponse_INVALID_REQUEST,
|
||||
Message: "encoding identifier is empty or exceeds the maximum length",
|
||||
Request: &livekit.RequestResponse_PublishDataTrack{
|
||||
PublishDataTrack: utils.CloneProto(req),
|
||||
},
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
publishedDataTracks := p.UpDataTrackManager.GetPublishedDataTracks()
|
||||
for _, dt := range publishedDataTracks {
|
||||
message := ""
|
||||
@@ -95,6 +108,8 @@ func (p *ParticipantImpl) HandlePublishDataTrackRequest(req *livekit.PublishData
|
||||
Name: req.Name,
|
||||
Encryption: req.Encryption,
|
||||
}
|
||||
dti.FrameEncoding = utils.CloneProto(req.GetFrameEncoding())
|
||||
dti.Schema = utils.CloneProto(req.GetSchema())
|
||||
dt := NewDataTrack(
|
||||
DataTrackParams{
|
||||
Logger: p.params.Logger.WithValues("trackID", dti.Sid),
|
||||
|
||||
@@ -414,6 +414,7 @@ type LocalParticipant interface {
|
||||
IsReady() bool
|
||||
ActiveAt() time.Time
|
||||
Disconnected() <-chan struct{}
|
||||
IsConnectionCanceled(closeReason ParticipantCloseReason) bool
|
||||
IsIdle() bool
|
||||
SubscriberAsPrimary() bool
|
||||
GetClientInfo() *livekit.ClientInfo
|
||||
|
||||
@@ -841,6 +841,17 @@ type FakeLocalParticipant struct {
|
||||
isClosedReturnsOnCall map[int]struct {
|
||||
result1 bool
|
||||
}
|
||||
IsConnectionCanceledStub func(types.ParticipantCloseReason) bool
|
||||
isConnectionCanceledMutex sync.RWMutex
|
||||
isConnectionCanceledArgsForCall []struct {
|
||||
arg1 types.ParticipantCloseReason
|
||||
}
|
||||
isConnectionCanceledReturns struct {
|
||||
result1 bool
|
||||
}
|
||||
isConnectionCanceledReturnsOnCall map[int]struct {
|
||||
result1 bool
|
||||
}
|
||||
IsDependentStub func() bool
|
||||
isDependentMutex sync.RWMutex
|
||||
isDependentArgsForCall []struct {
|
||||
@@ -5946,6 +5957,67 @@ func (fake *FakeLocalParticipant) IsClosedReturnsOnCall(i int, result1 bool) {
|
||||
}{result1}
|
||||
}
|
||||
|
||||
func (fake *FakeLocalParticipant) IsConnectionCanceled(arg1 types.ParticipantCloseReason) bool {
|
||||
fake.isConnectionCanceledMutex.Lock()
|
||||
ret, specificReturn := fake.isConnectionCanceledReturnsOnCall[len(fake.isConnectionCanceledArgsForCall)]
|
||||
fake.isConnectionCanceledArgsForCall = append(fake.isConnectionCanceledArgsForCall, struct {
|
||||
arg1 types.ParticipantCloseReason
|
||||
}{arg1})
|
||||
stub := fake.IsConnectionCanceledStub
|
||||
fakeReturns := fake.isConnectionCanceledReturns
|
||||
fake.recordInvocation("IsConnectionCanceled", []interface{}{arg1})
|
||||
fake.isConnectionCanceledMutex.Unlock()
|
||||
if stub != nil {
|
||||
return stub(arg1)
|
||||
}
|
||||
if specificReturn {
|
||||
return ret.result1
|
||||
}
|
||||
return fakeReturns.result1
|
||||
}
|
||||
|
||||
func (fake *FakeLocalParticipant) IsConnectionCanceledCallCount() int {
|
||||
fake.isConnectionCanceledMutex.RLock()
|
||||
defer fake.isConnectionCanceledMutex.RUnlock()
|
||||
return len(fake.isConnectionCanceledArgsForCall)
|
||||
}
|
||||
|
||||
func (fake *FakeLocalParticipant) IsConnectionCanceledCalls(stub func(types.ParticipantCloseReason) bool) {
|
||||
fake.isConnectionCanceledMutex.Lock()
|
||||
defer fake.isConnectionCanceledMutex.Unlock()
|
||||
fake.IsConnectionCanceledStub = stub
|
||||
}
|
||||
|
||||
func (fake *FakeLocalParticipant) IsConnectionCanceledArgsForCall(i int) types.ParticipantCloseReason {
|
||||
fake.isConnectionCanceledMutex.RLock()
|
||||
defer fake.isConnectionCanceledMutex.RUnlock()
|
||||
argsForCall := fake.isConnectionCanceledArgsForCall[i]
|
||||
return argsForCall.arg1
|
||||
}
|
||||
|
||||
func (fake *FakeLocalParticipant) IsConnectionCanceledReturns(result1 bool) {
|
||||
fake.isConnectionCanceledMutex.Lock()
|
||||
defer fake.isConnectionCanceledMutex.Unlock()
|
||||
fake.IsConnectionCanceledStub = nil
|
||||
fake.isConnectionCanceledReturns = struct {
|
||||
result1 bool
|
||||
}{result1}
|
||||
}
|
||||
|
||||
func (fake *FakeLocalParticipant) IsConnectionCanceledReturnsOnCall(i int, result1 bool) {
|
||||
fake.isConnectionCanceledMutex.Lock()
|
||||
defer fake.isConnectionCanceledMutex.Unlock()
|
||||
fake.IsConnectionCanceledStub = nil
|
||||
if fake.isConnectionCanceledReturnsOnCall == nil {
|
||||
fake.isConnectionCanceledReturnsOnCall = make(map[int]struct {
|
||||
result1 bool
|
||||
})
|
||||
}
|
||||
fake.isConnectionCanceledReturnsOnCall[i] = struct {
|
||||
result1 bool
|
||||
}{result1}
|
||||
}
|
||||
|
||||
func (fake *FakeLocalParticipant) IsDependent() bool {
|
||||
fake.isDependentMutex.Lock()
|
||||
ret, specificReturn := fake.isDependentReturnsOnCall[len(fake.isDependentArgsForCall)]
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
|
||||
"github.com/livekit/livekit-server/pkg/routing"
|
||||
"github.com/livekit/livekit-server/pkg/rtc/types"
|
||||
"github.com/livekit/livekit-server/pkg/sfu/rtpstats"
|
||||
"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
|
||||
"github.com/livekit/protocol/livekit"
|
||||
"github.com/livekit/protocol/logger"
|
||||
@@ -19,9 +20,8 @@ import (
|
||||
"github.com/livekit/psrpc"
|
||||
)
|
||||
|
||||
const (
|
||||
whipSessionNotifyInterval = 10 * time.Second
|
||||
)
|
||||
// whipSessionNotifyInterval is a var (rather than a const) so tests can shorten it.
|
||||
var whipSessionNotifyInterval = 10 * time.Second
|
||||
|
||||
type whipService struct {
|
||||
*RoomManager
|
||||
@@ -180,7 +180,7 @@ func (s whipService) notifySession(ctx context.Context, participant types.Partic
|
||||
case <-ticker.C:
|
||||
err := s.sendConnectionNotify(ctx, participant)
|
||||
if err != nil {
|
||||
if errors.Is(err, context.Canceled) {
|
||||
if errors.Is(err, context.Canceled) || errors.Is(err, ErrParticipantNotFound) {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
@@ -192,6 +192,10 @@ func (s whipService) notifySession(ctx context.Context, participant types.Partic
|
||||
}
|
||||
|
||||
func (s whipService) sendConnectionNotify(ctx context.Context, participant types.Participant) error {
|
||||
if participant.IsClosed() {
|
||||
return ErrParticipantNotFound
|
||||
}
|
||||
|
||||
video, audio := getMediaStateForParticipant(participant)
|
||||
|
||||
_, err := s.ingressRpcCli.WHIPRTCConnectionNotify(ctx, string(participant.ID()), &rpc.WHIPRTCConnectionNotifyRequest{
|
||||
@@ -204,52 +208,73 @@ func (s whipService) sendConnectionNotify(ctx context.Context, participant types
|
||||
}
|
||||
|
||||
func getMediaStateForParticipant(participant types.Participant) (*livekit.InputVideoState, *livekit.InputAudioState) {
|
||||
pParticipant := participant.ToProto()
|
||||
|
||||
var video *livekit.InputVideoState
|
||||
var audio *livekit.InputAudioState
|
||||
|
||||
for _, v := range pParticipant.Tracks {
|
||||
if v == nil {
|
||||
for _, t := range participant.GetPublishedTracks() {
|
||||
if t == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
if v.Type != livekit.TrackType_VIDEO {
|
||||
ti := t.ToProto()
|
||||
if ti == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
video = &livekit.InputVideoState{}
|
||||
switch t.Kind() {
|
||||
case livekit.TrackType_VIDEO:
|
||||
if video != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
video.MimeType = v.MimeType
|
||||
video.Height = v.Height
|
||||
video.Width = v.Width
|
||||
video = &livekit.InputVideoState{
|
||||
MimeType: ti.MimeType,
|
||||
Width: ti.Width,
|
||||
Height: ti.Height,
|
||||
AverageBitrate: trackAverageBitrate(t),
|
||||
}
|
||||
|
||||
break
|
||||
}
|
||||
case livekit.TrackType_AUDIO:
|
||||
if audio != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
for _, a := range pParticipant.Tracks {
|
||||
if a == nil {
|
||||
continue
|
||||
channels := uint32(1)
|
||||
if ti.Stereo {
|
||||
channels = 2
|
||||
}
|
||||
|
||||
audio = &livekit.InputAudioState{
|
||||
MimeType: ti.MimeType,
|
||||
Channels: channels,
|
||||
AverageBitrate: trackAverageBitrate(t),
|
||||
}
|
||||
}
|
||||
|
||||
if a.Type != livekit.TrackType_AUDIO {
|
||||
continue
|
||||
}
|
||||
|
||||
audio = &livekit.InputAudioState{}
|
||||
|
||||
audio.MimeType = a.MimeType
|
||||
audio.Channels = 1
|
||||
if a.Stereo {
|
||||
audio.Channels = 2
|
||||
}
|
||||
|
||||
break
|
||||
}
|
||||
|
||||
return video, audio
|
||||
}
|
||||
|
||||
func trackAverageBitrate(t types.MediaTrack) uint32 {
|
||||
var allStats []*livekit.RTPStats
|
||||
for _, r := range t.Receivers() {
|
||||
if r == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
if s := r.GetTrackStats(); s != nil {
|
||||
allStats = append(allStats, s)
|
||||
}
|
||||
}
|
||||
|
||||
agg := rtpstats.AggregateRTPStats(allStats)
|
||||
if agg == nil {
|
||||
return 0
|
||||
}
|
||||
|
||||
return uint32(agg.Bitrate)
|
||||
}
|
||||
|
||||
// -------------------------------------------
|
||||
|
||||
type whipParticipantService struct {
|
||||
@@ -323,12 +348,12 @@ func (r whipParticipantService) DeleteSession(ctx context.Context, req *rpc.WHIP
|
||||
|
||||
lp := room.GetParticipantByID(livekit.ParticipantID(req.ParticipantId))
|
||||
if lp != nil {
|
||||
lp.AddOnClose(types.ParticipantCloseKeyWHIP, nil)
|
||||
room.RemoveParticipant(
|
||||
lp.Identity(),
|
||||
lp.ID(),
|
||||
types.ParticipantCloseReasonClientRequestLeave,
|
||||
)
|
||||
lp.AddOnClose(types.ParticipantCloseKeyWHIP, nil)
|
||||
}
|
||||
|
||||
return &emptypb.Empty{}, nil
|
||||
|
||||
@@ -0,0 +1,125 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.uber.org/atomic"
|
||||
"google.golang.org/protobuf/types/known/emptypb"
|
||||
|
||||
"github.com/livekit/protocol/livekit"
|
||||
"github.com/livekit/protocol/rpc"
|
||||
"github.com/livekit/psrpc"
|
||||
|
||||
"github.com/livekit/livekit-server/pkg/rtc/types/typesfakes"
|
||||
)
|
||||
|
||||
// fakeIngressHandlerClient records WHIPRTCConnectionNotify calls. It embeds the
|
||||
// interface so only the method under test needs to be implemented; any other
|
||||
// call would panic (and we assert none happen).
|
||||
type fakeIngressHandlerClient struct {
|
||||
rpc.IngressHandlerClient
|
||||
notifyCount atomic.Int32
|
||||
}
|
||||
|
||||
func (f *fakeIngressHandlerClient) WHIPRTCConnectionNotify(
|
||||
_ context.Context,
|
||||
_ string,
|
||||
_ *rpc.WHIPRTCConnectionNotifyRequest,
|
||||
_ ...psrpc.RequestOption,
|
||||
) (*emptypb.Empty, error) {
|
||||
f.notifyCount.Inc()
|
||||
return &emptypb.Empty{}, nil
|
||||
}
|
||||
|
||||
// TestWhipNotifySessionStopsWhenParticipantLeaves verifies the notifier loop
|
||||
// terminates once the WHIP participant leaves the room (i.e. IsClosed becomes
|
||||
// true), and stops issuing further connection notifications.
|
||||
func TestWhipNotifySessionStopsWhenParticipantLeaves(t *testing.T) {
|
||||
origInterval := whipSessionNotifyInterval
|
||||
whipSessionNotifyInterval = 5 * time.Millisecond
|
||||
t.Cleanup(func() { whipSessionNotifyInterval = origInterval })
|
||||
|
||||
var closed atomic.Bool
|
||||
participant := &typesfakes.FakeParticipant{}
|
||||
participant.IsClosedStub = func() bool { return closed.Load() }
|
||||
participant.IDReturns(livekit.ParticipantID("PA_test"))
|
||||
participant.ToProtoReturns(&livekit.ParticipantInfo{})
|
||||
|
||||
cli := &fakeIngressHandlerClient{}
|
||||
s := whipService{ingressRpcCli: cli}
|
||||
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
done <- s.notifySession(context.Background(), participant)
|
||||
}()
|
||||
|
||||
// while the participant is connected the loop should keep notifying
|
||||
require.Eventually(t, func() bool {
|
||||
return cli.notifyCount.Load() > 0
|
||||
}, time.Second, time.Millisecond, "expected notifications while participant is connected")
|
||||
|
||||
// the participant leaves the room
|
||||
closed.Store(true)
|
||||
|
||||
select {
|
||||
case err := <-done:
|
||||
require.NoError(t, err)
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("notifySession did not stop after the participant left the room")
|
||||
}
|
||||
|
||||
// no further notifications should be attempted after it stops
|
||||
countAtStop := cli.notifyCount.Load()
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
require.Equal(t, countAtStop, cli.notifyCount.Load(), "should not notify after the participant left")
|
||||
}
|
||||
|
||||
// TestWhipNotifySessionStopsOnContextCancel verifies the loop exits when the
|
||||
// aliveCtx (cancelled from the participant's OnClose callback) is done.
|
||||
func TestWhipNotifySessionStopsOnContextCancel(t *testing.T) {
|
||||
origInterval := whipSessionNotifyInterval
|
||||
whipSessionNotifyInterval = 5 * time.Millisecond
|
||||
t.Cleanup(func() { whipSessionNotifyInterval = origInterval })
|
||||
|
||||
participant := &typesfakes.FakeParticipant{}
|
||||
participant.IsClosedReturns(false)
|
||||
participant.IDReturns(livekit.ParticipantID("PA_test"))
|
||||
participant.ToProtoReturns(&livekit.ParticipantInfo{})
|
||||
|
||||
cli := &fakeIngressHandlerClient{}
|
||||
s := whipService{ingressRpcCli: cli}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
done <- s.notifySession(ctx, participant)
|
||||
}()
|
||||
|
||||
cancel()
|
||||
|
||||
select {
|
||||
case err := <-done:
|
||||
require.NoError(t, err)
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("notifySession did not stop after context was cancelled")
|
||||
}
|
||||
}
|
||||
|
||||
// TestWhipSendConnectionNotifySkipsClosedParticipant verifies the guard that
|
||||
// short-circuits the RPC (and drives loop termination) for a closed participant.
|
||||
func TestWhipSendConnectionNotifySkipsClosedParticipant(t *testing.T) {
|
||||
participant := &typesfakes.FakeParticipant{}
|
||||
participant.IsClosedReturns(true)
|
||||
participant.IDReturns(livekit.ParticipantID("PA_test"))
|
||||
participant.ToProtoReturns(&livekit.ParticipantInfo{})
|
||||
|
||||
cli := &fakeIngressHandlerClient{}
|
||||
s := whipService{ingressRpcCli: cli}
|
||||
|
||||
err := s.sendConnectionNotify(context.Background(), participant)
|
||||
require.ErrorIs(t, err, ErrParticipantNotFound)
|
||||
require.Zero(t, cli.notifyCount.Load(), "should not issue an RPC for a closed participant")
|
||||
}
|
||||
+1
-1
@@ -14,4 +14,4 @@
|
||||
|
||||
package version
|
||||
|
||||
const Version = "1.13.1"
|
||||
const Version = "1.13.3"
|
||||
|
||||
Reference in New Issue
Block a user