DOCSEARCHbyAlgolia
Submit Docs

LiveKit Ingress

repository·main·Indexed 41 days ago

https://github.com/livekit/ingress

A service that enables media ingestion from non-WebRTC sources, such as RTMP and WHIP (WebRTC-HTTP Ingestion Protocol), into LiveKit rooms by automatically transcoding them for WebRTC compatibility. It supports push and pull models, simulcast, and configurable video layers.

Tokens
5.2K
Snippets
15
Records
26
Agent score
52%

What's inside LiveKit Ingress

  1. Overview of LiveKit Ingress

    main
    LiveKit Ingress (Universal Ingress) is a service designed to bridge non-WebRTC media workflows into a LiveKit room. It allows you to ingest media from equipment or workflows that do not natively support WebRTC, making it a versatile tool for integrating existing media infrastructure with LiveKit's WebRTC-based delivery.

  3. How the Ingress push workflow works

    main

    To push media to LiveKit Ingress, follow these steps:

    1. Create an Ingress: Call the CreateIngress API on the LiveKit server.
    2. Get the URL: The API returns a unique URL for the session.
    3. Configure Encoder: Copy the returned URL into your streaming software (e.g., OBS) or workflow.
    4. Start Stream: Once the stream starts, the Ingress service receives the data.
    5. Publish to Room: The Ingress service joins the specified LiveKit room and publishes the transcoded media as a participant.

  4. Supported Output and Transcoding

    main

    The Ingress service automatically transcodes source media to ensure full compatibility with WebRTC.

    Key output features include:

    • Simulcast Support: The service can publish multiple video layers to optimize delivery for different participants.
    • Layer Configuration: The parameters for different video layers can be defined at the time the ingress is created.

  5. Run Ingress service natively

    main

    The Ingress service is built in Go and requires GStreamer.

    Prerequisites

    • Go >= 1.18
    • GStreamer libraries: gstreamer, gst-plugins-base, gst-plugins-good, gst-plugins-bad, gst-plugins-ugly, gst-libav.

    Build and Run

    1. Build: mage build
    2. Run: ingress --config=config.yaml

    MacOS Note: If using Homebrew for GStreamer, you must set the library path: export DYLD_LIBRARY_PATH=/opt/homebrew/lib

    Troubleshooting: If you encounter GStreamer plugin errors (e.g., Failed to load libsoup library), clear the registry: rm -rf ~/.cache/gstreamer-1.0/

    # Build
mage build

# Run
export DYLD_LIBRARY_PATH=/opt/homebrew/lib
ingress --config=config.yaml

  6. How RTMPHandler processes media

    main

    The RTMPHandler is the internal component responsible for handling individual RTMP connections. It implements the rtmp.DefaultHandler interface and performs the following:

    1. Publishing: When a client calls publish, the handler triggers the onPublish callback. This is where you define the stream's parameters and initialize statistics gathering.
    2. FLV Encoding: The handler automatically initializes an FLV encoder (flv.Encoder) upon receiving the first data frames. It uses a PrerollBuffer to ensure that if a relay is attached later, the initial metadata (like video/audio init tags) can be replayed.
    3. Media Handling:
      • OnAudio and OnVideo receive raw payloads, decode them into FLV tags, and encode them into the FLV stream.
      • It tracks keyframes to ensure the stream starts correctly.
      • It updates media statistics via the registered trackStats.
    4. Lifecycle: When the connection closes, OnClose is called, which closes the mediaBuffer and triggers the onClose callback.

  7. Configure the Ingress service via YAML

    main

    The Ingress service is configured using a YAML file. You can provide this file via the INGRESS_CONFIG_FILE environment variable or pass the raw YAML content via INGRESS_CONFIG_BODY.

    Required Fields

    • api_key: LiveKit server API key (or use LIVEKIT_API_KEY env).
    • api_secret: LiveKit server API secret (or use LIVEKIT_API_SECRET env).
    • ws_url: LiveKit server websocket URL (or use LIVEKIT_WS_URL env).
    • redis: Configuration for the Redis instance used for communication and state.
      • address: Redis address.
      • username: Redis username.
      • password: Redis password.
      • db: Redis database index.

    Common Optional Fields

    • health_port: HTTP port for health checks.
    • prometheus_port: Port for Prometheus metrics.
    • log_level: debug, info, warn, or error (default: info).
    • rtmp_port: Port for RTMP connections (default: 1935).
    • whip_port: Port for WHIP calls (default: 8080).
    • http_relay_port: Port for data relay between main and handler processes (default: 9090).
    • rtc_config: RTC/ICE settings for WHIP.
    • enable_udp_url_pull: Enables pulling from udp:// URLs (default: false).
    • psrpc.compression.quality: Gzip level 1-9 (0 to disable, default: 0).
    # Example minimal config
api_key: livekit server api key
api_secret: livekit server api secret
ws_url: livekit server websocket url
redis:
  address: redis address
  username: redis username
  password: redis password
  db: 0

  8. How the Ingress service handles shutdown

    main

    The Ingress server responds to OS signals to manage the lifecycle of active media streams:

    • SIGTERM / SIGQUIT: The server attempts a graceful shutdown by finishing all active ingresses before shutting down.
    • SIGINT (Ctrl+C): The server performs an immediate shutdown, stopping all ingresses, the relay, the RTMP server, and the WHIP server.

    This ensures that media streams are either allowed to complete or are terminated cleanly to prevent resource leaks.

  9. Configure LiveKit Server for Ingress

    main

    For the LiveKit server to successfully create Ingress sessions, you must add an ingress section to your livekit-server configuration. This defines the base URLs that the server will use to construct the final publishing endpoints (by appending a stream key to these prefixes).

    Required keys:

    • rtmp_base_url: The RTMP URL prefix pointing to your Ingress external IP or load balancer.
    • whip_base_url: The WHIP URL prefix pointing to your Ingress external IP or load balancer.
    ingress:
  rtmp_base_url: rtmp://my.domain.com/x
  whip_base_url: http://my.domain.com/w

  10. Run the RTMP relay service

    main

    The rtmprelay service is a CLI entrypoint used to start an RTMP server and a relay service. It listens for RTMP connections and relays them through the LiveKit Ingress system.

    By default, the service initializes with:

    • RTMPPort: 1935
    • HTTPRelayPort: 9090

    The service handles graceful shutdowns when receiving a SIGINT signal.

    // The service initializes an RTMP server and a relay service
// using the following default configuration:
conf := &config.Config{
	RTMPPort:      1935,
	HTTPRelayPort: 9090,
}

rtmpServer := rtmp.NewRTMPServer()
relay := service.NewRelay(rtmpServer)

// Start the components
rtmpServer.Start(conf, nil)
relay.Start(conf)

  11. Run the LiveKit Ingress server

    main

    The ingress CLI is used to import streamed media to LiveKit. You can start the main service by providing a configuration file or a raw YAML configuration body.

    Configuration Options

    • --config: Path to a LiveKit Ingress YAML configuration file. Can be set via the INGRESS_CONFIG_FILE environment variable.
    • --config-body: A raw YAML configuration string. Can be set via the INGRESS_CONFIG_BODY environment variable.

    When running the service, it initializes components like the Redis message bus, RTMP server (if RTMPPort is configured), and WHIP server (if WHIPPort is configured).

    # Using a config file
./ingress --config /path/to/config.yaml

# Using a config body
./ingress --config-body "redis: { address: 'localhost:6379' }"

  12. Initialize and start a WHIPServer

    main

    To use the WHIP (WebRTC-HTTP Ingestion Protocol) server, you must first create it using NewWHIPServer with a psrpc.MessageBus. Then, call Start with your configuration, an onPublish callback, and optional health handlers. The onPublish callback is critical as it defines how new streams are provisioned.

    The onPublish callback signature: It takes streamKey and resourceId (strings) and returns:

    1. *params.Params: The parameters for the new stream.
    2. ready function: Called when the session is ready. It accepts mimeTypes and an error, and returns a *stats.LocalMediaStatsGatherer and a cleanup function.
    3. ended function: A cleanup function called when the session ends.
    4. error: Any error encountered during provisioning.
    // Initialize the server
server, err := whip.NewWHIPServer(bus)
if err != nil {
	return err
}

// Define the onPublish callback
onPublish := func(streamKey, resourceId string) (*params.Params, func(mimeTypes map[types.StreamKind]string, err error) *stats.LocalMediaStatsGatherer, func(error), error) {
	// Logic to provision the stream
	return params, readyFunc, endedFunc, nil
}

// Start the server
err = server.Start(conf, onPublish, healthHandlers)