Skip to navigation
Video

Manage streams

Preview

List, retrieve, stop, and re-egress existing video streams.

Use the Lattice SDK to inspect active streams, retrieve a specific stream, archive a stream when its source goes offline, and re-publish over RTSP or SRT for downstream consumers.

Before you begin

gRPC authentication

These gRPC examples authenticate with a static environment token attached as request metadata. If you are using OAuth 2.0 client credentials instead, set up the token refresh module before running the examples on this page.

Stream ownership

Your integration owns the lifecycle of the IngressStream records it creates. Streams are not automatically removed when a publisher disconnects, a stream becomes inactive, or your integration stops communicating with Lattice. A stream with status STREAM_STATUS_INACTIVE still exists; when the source or publisher reconnects, reuse the existing stream rather than creating another one.

Persist each stream’s ingress_id with the asset or integration that owns it. On startup or reconnect, call GetIngressStream with the saved ID to check the stream. If you don’t have the ID, use ListIngressStreams and inspect its pages to find the stream. Create a new stream only after checking that an existing one for the source does not already exist.

List active streams

Call ListIngressStreams to retrieve the current page of active streams. Archived streams are not returned. Send an empty request to fetch the first page (default: 50, max: 100), then iterate while next_page_token is non-empty.

Pagination contract

Pass the previous response’s next_page_token as page_token and keep every other request field identical. The server may reject requests that change other fields between pages.

// This Go example uses the Lattice REST SDK (lattice-sdk-go) to list active ingress streams.
package main
import (
	"context"
	"fmt"
	"net/http"
	"os"
	Lattice "github.com/anduril/lattice-sdk-go/v5"
	"github.com/anduril/lattice-sdk-go/v5/client"
	"github.com/anduril/lattice-sdk-go/v5/option"
)
func main() {
	// Get environment variables
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")
	// Check required environment variables
	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}
	// Initialize headers for sandbox authorization
	headers := http.Header{}
	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
	// Create the client
	LatticeClient := client.NewClient(
		option.WithToken(environmentToken),
		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
		option.WithHTTPHeader(headers),
	)
	ctx := context.Background()
	// Iterate every page of active streams. Keep all request fields identical between
	// pages other than pageToken; the server may reject a request that changes them.
	pageToken := ""
	for {
		response, err := LatticeClient.Video.ListIngressStreams(ctx, &Lattice.ListIngressStreamsRequest{
			PageToken: &pageToken,
		})
		if err != nil {
			fmt.Printf("Failed to list streams: %v\n", err)
			os.Exit(1)
		}
		for _, stream := range response.GetIngressStreams() {
			fmt.Printf("Stream %s status %s\n", *stream.GetIngressID(), *stream.GetStatus())
		}
		if response.GetNextPageToken() == nil || *response.GetNextPageToken() == "" {
			break
		}
		pageToken = *response.GetNextPageToken()
	}
}

Get a single stream

Call GetIngressStream with an ingress_id to retrieve a single IngressStream. The response includes the current status, ingress configuration, and associated egress stream identifiers. For a description of every field, see IngressStream object.

MPEG-TS streams are read-only over the public internet

MPEG-TS ingress is supported only at the edge, in closed networks. If you connect to Lattice over the public internet, you can’t start an MPEG-TS stream, but a stream created at the edge still appears in ListIngressStreams and can be inspected with GetIngressStream.

// This Go example uses the Lattice REST SDK (lattice-sdk-go) to retrieve a single ingress stream.
package main
import (
	"context"
	"fmt"
	"net/http"
	"os"
	Lattice "github.com/anduril/lattice-sdk-go/v5"
	"github.com/anduril/lattice-sdk-go/v5/client"
	"github.com/anduril/lattice-sdk-go/v5/option"
)
// Ingress ID of the stream to get. Replace before running.
const ingressID = "<INGRESS-ID>"
func main() {
	// Get environment variables
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")
	// Check required environment variables
	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}
	// Initialize headers for sandbox authorization
	headers := http.Header{}
	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
	// Create the client
	LatticeClient := client.NewClient(
		option.WithToken(environmentToken),
		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
		option.WithHTTPHeader(headers),
	)
	ctx := context.Background()
	response, err := LatticeClient.Video.GetIngressStream(ctx, &Lattice.GetIngressStreamRequest{
		IngressID: ingressID,
	})
	if err != nil {
		fmt.Printf("Failed to get stream: %v\n", err)
		os.Exit(1)
	}
	stream := response.GetIngressStream()
	if stream == nil {
		fmt.Printf("No ingress stream found for %s\n", ingressID)
		os.Exit(1)
	}
	fmt.Printf("Ingress ID: %s\n", *stream.GetIngressID())
	fmt.Printf("Title:      %q\n", *stream.GetTitle())
	fmt.Printf("Status:     %s\n", *stream.GetStatus())
}

Stop an ingress stream

Stopping a stream archives it. The IngressStream record remains retrievable with GetIngressStream, but its status transitions to STREAM_STATUS_ARCHIVED and ListIngressStreams no longer returns the record.

Archive, not delete

DeleteIngressStream does not delete the underlying record. To remove the stale media reference from the camera entity, follow up with an OverrideEntity call as shown in the example below. The example reads the entity, filters out the stale MediaItem, and writes the surviving items back. A field_path of media replaces the entire Media component while leaving every other entity component untouched.

// This Go example uses the Lattice REST SDK (lattice-sdk-go) to stop (archive) an ingress stream
// and remove its stale media reference from the associated camera entity.
package main
import (
	"context"
	"fmt"
	"net/http"
	"os"
	"time"
	Lattice "github.com/anduril/lattice-sdk-go/v5"
	"github.com/anduril/lattice-sdk-go/v5/client"
	"github.com/anduril/lattice-sdk-go/v5/option"
)
const (
	// Ingress ID of the stream to stop. Replace before running.
	ingressID = "<DELETE-INGRESS-ID>"
	// Entity the video is attached to. Set to "" to only stop the stream without
	// editing any entity's media list.
	entityID = "<ENTITY-ID>"
)
func main() {
	// Get environment variables
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")
	// Check required environment variables
	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}
	// Initialize headers for sandbox authorization
	headers := http.Header{}
	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
	// Create the client
	LatticeClient := client.NewClient(
		option.WithToken(environmentToken),
		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
		option.WithHTTPHeader(headers),
	)
	ctx := context.Background()
	fmt.Printf("Stopping ingress stream with ID: %s\n", ingressID)
	_, err := LatticeClient.Video.DeleteIngressStream(ctx, &Lattice.DeleteIngressStreamRequest{
		IngressID: ingressID,
	})
	if err != nil {
		fmt.Printf("Failed to stop ingress stream: %v\n", err)
		os.Exit(1)
	}
	// With no entity to update, stopping the stream is all there is to do.
	if entityID == "" {
		return
	}
	// Read the entity, filter the stale MediaItem out, and write the surviving items back.
	entity, err := LatticeClient.Entities.GetEntity(ctx, &Lattice.GetEntityRequest{
		EntityID: entityID,
	})
	if err != nil {
		fmt.Printf("Failed to fetch entity %s: %v\n", entityID, err)
		os.Exit(1)
	}
	var current []*Lattice.MediaItem
	if entity.GetMedia() != nil {
		current = entity.GetMedia().GetMedia()
	}
	remaining := make([]*Lattice.MediaItem, 0, len(current))
	for _, item := range current {
		if item.GetItemIdentifier() == nil || *item.GetItemIdentifier() != ingressID {
			remaining = append(remaining, item)
		}
	}
	if len(remaining) == len(current) {
		fmt.Printf("Video %s was not in entity %s media list; nothing to remove\n", ingressID, entityID)
		return
	}
	// A field_path of media replaces the entire Media component with the surviving items.
	_, err = LatticeClient.Entities.OverrideEntity(ctx, &Lattice.EntityOverride{
		EntityID:  entityID,
		FieldPath: "media",
		Entity: &Lattice.Entity{
			EntityID: Lattice.String(entityID),
			Media: &Lattice.Media{
				Media: remaining,
			},
		},
		Provenance: &Lattice.Provenance{
			IntegrationName:  Lattice.String("lattice-video-integration"),
			DataType:         Lattice.String("video-association"),
			SourceUpdateTime: Lattice.Time(time.Now().UTC()),
		},
	})
	if err != nil {
		fmt.Printf("Failed to override entity media: %v\n", err)
		os.Exit(1)
	}
	fmt.Printf("Removed video %s from entity %s\n", ingressID, entityID)
}

List active egress streams

Call ListEgressStreams to retrieve the current page of active egress streams. Each entry pairs an egress_id with the ingress_id it re-publishes, so you can see every downstream consumer of a source. Send an empty request to fetch the first page (default: 50, max: 100), then iterate while next_page_token is non-empty.

Pagination contract

Pass the previous response’s next_page_token as page_token and keep every other request field identical. The server may reject requests that change other fields between pages.

// This Go example uses the Lattice REST SDK (lattice-sdk-go) to list active egress streams.
package main
import (
	"context"
	"fmt"
	"net/http"
	"os"
	Lattice "github.com/anduril/lattice-sdk-go/v5"
	"github.com/anduril/lattice-sdk-go/v5/client"
	"github.com/anduril/lattice-sdk-go/v5/option"
)
func main() {
	// Get environment variables
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")
	// Check required environment variables
	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}
	// Initialize headers for sandbox authorization
	headers := http.Header{}
	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
	// Create the client
	LatticeClient := client.NewClient(
		option.WithToken(environmentToken),
		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
		option.WithHTTPHeader(headers),
	)
	ctx := context.Background()
	// Iterate every page of active streams. Keep all request fields identical between
	// pages other than pageToken; the server may reject a request that changes them.
	pageToken := ""
	for {
		response, err := LatticeClient.Video.ListEgressStreams(ctx, &Lattice.ListEgressStreamsRequest{
			PageToken: &pageToken,
		})
		if err != nil {
			fmt.Printf("Failed to list egress streams: %v\n", err)
			os.Exit(1)
		}
		for _, stream := range response.GetEgressStreams() {
			fmt.Printf("Egress %s re-publishes ingress %s\n", *stream.GetEgressID(), *stream.GetIngressID())
		}
		if response.GetNextPageToken() == nil || *response.GetNextPageToken() == "" {
			break
		}
		pageToken = *response.GetNextPageToken()
	}
}

Get a single egress stream

Call GetEgressStream with an egress_id to retrieve a single EgressStream. The response includes the ingress_id it re-publishes and the protocol-specific connection details, either an RtspEgress or an SrtEgress, that a downstream consumer uses to pull the stream.

// This Go example uses the Lattice REST SDK (lattice-sdk-go) to retrieve a single egress stream.
package main
import (
	"context"
	"fmt"
	"net/http"
	"os"
	Lattice "github.com/anduril/lattice-sdk-go/v5"
	"github.com/anduril/lattice-sdk-go/v5/client"
	"github.com/anduril/lattice-sdk-go/v5/option"
)
// Egress ID of the stream to get. Replace before running.
const egressID = "<EGRESS-ID>"
func main() {
	// Get environment variables
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")
	// Check required environment variables
	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}
	// Initialize headers for sandbox authorization
	headers := http.Header{}
	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
	// Create the client
	LatticeClient := client.NewClient(
		option.WithToken(environmentToken),
		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
		option.WithHTTPHeader(headers),
	)
	ctx := context.Background()
	response, err := LatticeClient.Video.GetEgressStream(ctx, &Lattice.GetEgressStreamRequest{
		EgressID: egressID,
	})
	if err != nil {
		fmt.Printf("Failed to get egress stream: %v\n", err)
		os.Exit(1)
	}
	stream := response.GetEgressStream()
	if stream == nil {
		fmt.Printf("No egress stream found for %s\n", egressID)
		os.Exit(1)
	}
	fmt.Printf("Egress ID:  %s\n", *stream.GetEgressID())
	fmt.Printf("Ingress ID: %s\n", *stream.GetIngressID())
	// Exactly one of rtsp or srt is set, matching the transport the egress stream uses.
	if stream.GetRtsp() != nil {
		fmt.Printf("RTSP URL:   %s\n", *stream.GetRtsp().GetURL())
	} else if stream.GetSrt() != nil {
		fmt.Printf("SRT URL:    %s\n", *stream.GetSrt().GetURL())
		fmt.Printf("Session ID: %s\n", *stream.GetSrt().GetSessionID())
	}
}

Stop an egress stream

Call DeleteEgressStream with an egress_id to stop delivering a stream to its downstream consumers. Deleting an egress stream stops only that downstream delivery; the underlying ingress stream and any other egress streams consuming it are unaffected.

Egress streams stop with their source

Deleting an IngressStream automatically stops every egress stream consuming it, so you don’t need to delete egress streams individually before stopping their source. DeleteEgressStream returns NOT_FOUND if no matching active egress stream exists.

// This Go example uses the Lattice REST SDK (lattice-sdk-go) to stop an egress stream,
// which halts delivery to its downstream consumers.
package main
import (
	"context"
	"fmt"
	"net/http"
	"os"
	Lattice "github.com/anduril/lattice-sdk-go/v5"
	"github.com/anduril/lattice-sdk-go/v5/client"
	"github.com/anduril/lattice-sdk-go/v5/option"
)
// Egress ID of the stream to stop. Replace before running.
const egressID = "<DELETE-EGRESS-ID>"
func main() {
	// Get environment variables
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")
	// Check required environment variables
	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}
	// Initialize headers for sandbox authorization
	headers := http.Header{}
	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
	// Create the client
	LatticeClient := client.NewClient(
		option.WithToken(environmentToken),
		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
		option.WithHTTPHeader(headers),
	)
	ctx := context.Background()
	// Deleting an egress stream stops only that downstream delivery. The underlying ingress
	// stream and any other egress streams consuming it are unaffected. Deleting an egress
	// stream that does not exist returns NOT_FOUND.
	fmt.Printf("Stopping egress stream with ID: %s\n", egressID)
	_, err := LatticeClient.Video.DeleteEgressStream(ctx, &Lattice.DeleteEgressStreamRequest{
		EgressID: egressID,
	})
	if err != nil {
		fmt.Printf("Failed to stop egress stream: %v\n", err)
		os.Exit(1)
	}
	fmt.Printf("Stopped egress stream %s\n", egressID)
}

What’s next

  • Overview: Review the API surface and stream lifecycle.
  • Create streams: Publish an asset and start an ingress stream.