> This is the Markdown version of the page. For the complete Lattice Developers documentation index, fetch https://developer.anduril.com/llms.txt. Append .md to any page URL for its clean Markdown. # Manage streams > 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 * To inspect and manage streams, [set up](/guides/getting-started/set-up) your Lattice environment. * See an overview of the [Video Manager API and stream lifecycle](/guides/video/overview) in Lattice. * If you haven't already, learn how to [stream a video](/guides/video/create-streams). > **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](/guides/getting-started/authenticate#client-credentials) 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. ```go title={"Go (REST)"} // 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() } } ``` ```java title={"Java (REST)"} package org.example; import java.util.List; import com.anduril.Lattice; import com.anduril.resources.video.requests.ListIngressStreamsRequest; import com.anduril.types.IngressStream; import com.anduril.types.ListIngressStreamsResponse; /** * Lists active ingress streams with the Lattice SDK. */ public class ListIngressStreams { public static void main(String[] args) { // Get environment variables String endpoint = System.getenv("LATTICE_ENDPOINT"); String environmentToken = System.getenv("ENVIRONMENT_TOKEN"); String sandboxesToken = System.getenv("SANDBOXES_TOKEN"); // Check required variables and ensure URL has https:// if (endpoint == null || environmentToken == null || sandboxesToken == null) { System.err.println("Missing required environment variables"); System.exit(1); } if (!endpoint.startsWith("https://")) endpoint = "https://" + endpoint; try { // Create Lattice client with Sandbox authentication Lattice client = Lattice.builder() .url(endpoint) .token(environmentToken) .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken) .build(); // Iterate every page of active streams. Keep all request fields identical between // pages other than page_token; the server may reject a request that changes them. String pageToken = ""; while (true) { ListIngressStreamsResponse response = client.video().listIngressStreams( ListIngressStreamsRequest.builder() .pageToken(pageToken) .build()); List streams = response.getIngressStreams().orElse(List.of()); for (IngressStream stream : streams) { System.out.println("Stream " + stream.getIngressId().orElse("") + " status " + stream.getStatus().map(Object::toString).orElse("")); } String nextPageToken = response.getNextPageToken().orElse(""); if (nextPageToken.isEmpty()) { break; } pageToken = nextPageToken; } } catch (Exception e) { System.err.println("Failed to list streams: " + e.getMessage()); System.exit(1); } } } ``` ```py title={"Python (REST)"} # This Python example uses the Lattice REST SDK (anduril-lattice-sdk) to list active ingress # streams. import os import sys from anduril import Lattice def main(): environment_token = os.getenv("ENVIRONMENT_TOKEN") lattice_endpoint = os.getenv("LATTICE_ENDPOINT") # Remove sandboxes_token from the following statements if you are not developing on Sandboxes. sandboxes_token = os.getenv("SANDBOXES_TOKEN") if not environment_token or not lattice_endpoint or not sandboxes_token: print("Missing required environment variables.", file=sys.stderr) sys.exit(1) client = Lattice( base_url=f"https://{lattice_endpoint}", token=lambda: str(environment_token), # Remove the following header if you are not developing on Sandboxes. headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"}, ) # Iterate every page of active streams. Keep all request fields identical between # pages other than page_token; the server may reject a request that changes them. page_token = "" while True: try: response = client.video.list_ingress_streams(page_token=page_token) except Exception as error: print(f"Failed to list streams: {error}", file=sys.stderr) sys.exit(1) for stream in response.ingress_streams or []: print(f"Stream {stream.ingress_id} status {stream.status}") if not response.next_page_token: break page_token = response.next_page_token if __name__ == "__main__": main() ``` ```ts title={"Typescript (REST)"} import { LatticeClient } from "@anduril-industries/lattice-sdk"; const latticeEndpoint = process.env.LATTICE_ENDPOINT; const environmentToken = process.env.ENVIRONMENT_TOKEN; // Remove sandboxesToken from the following statements if you are not developing on Sandboxes. const sandboxesToken = process.env.SANDBOXES_TOKEN; if (!latticeEndpoint || !environmentToken || !sandboxesToken) { console.log('Missing required environment variables.'); process.exit(1); } const client = new LatticeClient( { baseUrl: `https://${latticeEndpoint}`, token: environmentToken, // Remove the following if you are not developing on Sandboxes. headers: { "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}` } } ); async function App() { try { // 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. let pageToken = ""; while (true) { const response = await client.video.listIngressStreams({ pageToken }); for (const stream of response.ingressStreams ?? []) { console.log(`Stream ${stream.ingressId} status ${stream.status}`); } if (!response.nextPageToken) { break; } pageToken = response.nextPageToken; } } catch (error) { console.log(`Encountered the following error while listing streams: ${error}`); } } App(); ``` ```go title={"Go (gRPC)"} // This Go example is compatible with artifacts generated using // the following grpc/go plugin: https://buf.build/anduril/lattice-sdk/sdks/main:grpc/go package main import ( "context" "fmt" "log" "os" "time" "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/videomanager/v1/videomanagerv1grpc" videomanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/videomanager/v1" "google.golang.org/grpc" "google.golang.org/grpc/credentials" ) func main() { environmentToken := os.Getenv("ENVIRONMENT_TOKEN") latticeEndpoint := os.Getenv("LATTICE_ENDPOINT") sandboxesToken := os.Getenv("SANDBOXES_TOKEN") if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" { log.Fatalf("Missing required environment variables") } auth := &BearerTokenAuth{ Token: environmentToken, SandboxesToken: sandboxesToken, } opts := []grpc.DialOption{ grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")), grpc.WithPerRPCCredentials(auth), } conn, err := grpc.NewClient(latticeEndpoint, opts...) if err != nil { log.Fatalf("Did not connect: %v", err) } defer conn.Close() client := videomanagerv1grpc.NewVideoManagerAPIClient(conn) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() // Iterate every page of active streams. Keep all request fields identical between // pages other than page_token; the server may reject a request that changes them. pageToken := "" for { resp, err := client.ListIngressStreams(ctx, &videomanagerv1.ListIngressStreamsRequest{ PageToken: pageToken, }) if err != nil { log.Fatalf("Failed to list streams: %v", err) } for _, stream := range resp.GetIngressStreams() { fmt.Printf("Stream %s status %s\n", stream.GetIngressId(), stream.GetStatus()) } if resp.GetNextPageToken() == "" { break } pageToken = resp.GetNextPageToken() } } ``` ```py title={"Python (gRPC)"} # This Python example is compatible with artifacts generated using the following # Buf plugins: https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:bufbuild/py # and https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:connectrpc/py import os import sys from bearer_auth import create_client from anduril.videomanager.v1.list_ingress_streams_request_pub_pb import ( ListIngressStreamsRequest, ) from anduril.videomanager.v1.video_manager_api_pub_connect import ( VideoManagerAPIClientSync, ) def main(): environment_token = os.getenv("ENVIRONMENT_TOKEN") lattice_endpoint = os.getenv("LATTICE_ENDPOINT") sandboxes_token = os.getenv("SANDBOXES_TOKEN") if not environment_token or not lattice_endpoint or not sandboxes_token: print("Missing required environment variables", file=sys.stderr) sys.exit(1) # Create the API client over Connect gRPC with authentication client = create_client(VideoManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token) # Iterate every page of active streams. Keep all request fields identical between # pages other than page_token; the server may reject a request that changes them. page_token = "" while True: try: response = client.list_ingress_streams( ListIngressStreamsRequest(page_token=page_token), timeout_ms=10000, ) except Exception as error: print(f"Failed to list streams: {error}", file=sys.stderr) sys.exit(1) for stream in response.ingress_streams: print(f"Stream {stream.ingress_id} status {stream.status}") if not response.next_page_token: break page_token = response.next_page_token if __name__ == "__main__": main() ``` ```rs title={"Rust (gRPC)"} // This Rust example is compatible with artifacts generated using // the following grpc/rust plugin: https://buf.build/anduril/lattice-sdk/sdks/main:community/neoeinstein-tonic mod bearer_auth; use anduril_lattice_sdk_community_neoeinstein_prost::anduril::videomanager::v1::ListIngressStreamsRequest; use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::videomanager::v1::tonic::video_manager_api_client::VideoManagerApiClient; use bearer_auth::EnvironmentTokenAuth; use tonic::metadata::MetadataValue; use tonic::transport::{Channel, ClientTlsConfig}; use tonic::Request; use std::env; #[tokio::main] async fn main() -> Result<(), Box> { let environment_token = env::var("ENVIRONMENT_TOKEN") .expect("ENVIRONMENT_TOKEN environment variable not set"); let lattice_endpoint = env::var("LATTICE_ENDPOINT") .expect("LATTICE_ENDPOINT environment variable not set"); let sandboxes_token = env::var("SANDBOXES_TOKEN") .expect("SANDBOXES_TOKEN environment variable not set"); let auth = EnvironmentTokenAuth::new( environment_token, sandboxes_token, ); let tls_config = ClientTlsConfig::new().with_native_roots(); let channel = Channel::from_shared(format!("https://{}", lattice_endpoint))? .tls_config(tls_config)? .connect() .await?; let access_token = auth.get_token().await?; let sandboxes_token = auth.sandboxes_token(); let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?; let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?; let mut client = VideoManagerApiClient::with_interceptor( channel, move |mut req: Request<()>| { req.metadata_mut() .insert("authorization", auth_header.clone()); req.metadata_mut() .insert("anduril-sandbox-authorization", sandbox_header.clone()); Ok(req) }, ); // Iterate every page of active streams. Keep all request fields identical between // pages other than page_token; the server may reject a request that changes them. let mut page_token = String::new(); loop { let request = Request::new(ListIngressStreamsRequest { page_size: 0, page_token: page_token.clone(), }); let response = client.list_ingress_streams(request).await?.into_inner(); for stream in &response.ingress_streams { println!( "Stream {} status {:?}", stream.ingress_id, stream.status() ); } if response.next_page_token.is_empty() { break; } page_token = response.next_page_token; } Ok(()) } ``` ## 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](/guides/video/overview#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`. ```go title={"Go (REST)"} // 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 = "" 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()) } ``` ```java title={"Java (REST)"} package org.example; import com.anduril.Lattice; import com.anduril.types.GetIngressStreamResponse; import com.anduril.types.IngressStream; /** * Retrieves a single ingress stream with the Lattice SDK. */ public class GetIngressStream { // Ingress ID of the stream to get. Replace before running. private static final String INGRESS_ID = ""; public static void main(String[] args) { // Get environment variables String endpoint = System.getenv("LATTICE_ENDPOINT"); String environmentToken = System.getenv("ENVIRONMENT_TOKEN"); String sandboxesToken = System.getenv("SANDBOXES_TOKEN"); // Check required variables and ensure URL has https:// if (endpoint == null || environmentToken == null || sandboxesToken == null) { System.err.println("Missing required environment variables"); System.exit(1); } if (!endpoint.startsWith("https://")) endpoint = "https://" + endpoint; try { // Create Lattice client with Sandbox authentication Lattice client = Lattice.builder() .url(endpoint) .token(environmentToken) .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken) .build(); GetIngressStreamResponse response = client.video().getIngressStream(INGRESS_ID); IngressStream stream = response.getIngressStream().orElse(null); if (stream == null) { System.err.println("No ingress stream found for " + INGRESS_ID); System.exit(1); } System.out.println("Ingress ID: " + stream.getIngressId().orElse("")); System.out.println("Title: " + stream.getTitle().orElse("")); System.out.println("Status: " + stream.getStatus().map(Object::toString).orElse("")); } catch (Exception e) { System.err.println("Failed to get stream: " + e.getMessage()); System.exit(1); } } } ``` ```py title={"Python (REST)"} # This Python example uses the Lattice REST SDK (anduril-lattice-sdk) to retrieve a single ingress # stream. import os import sys from anduril import Lattice # Ingress ID of the stream to get. Replace before running. INGRESS_ID = "" def main(): environment_token = os.getenv("ENVIRONMENT_TOKEN") lattice_endpoint = os.getenv("LATTICE_ENDPOINT") # Remove sandboxes_token from the following statements if you are not developing on Sandboxes. sandboxes_token = os.getenv("SANDBOXES_TOKEN") if not environment_token or not lattice_endpoint or not sandboxes_token: print("Missing required environment variables.", file=sys.stderr) sys.exit(1) client = Lattice( base_url=f"https://{lattice_endpoint}", token=lambda: str(environment_token), # Remove the following header if you are not developing on Sandboxes. headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"}, ) try: response = client.video.get_ingress_stream(INGRESS_ID) except Exception as error: print(f"Failed to get stream: {error}", file=sys.stderr) sys.exit(1) stream = response.ingress_stream if not stream: print(f"No ingress stream found for {INGRESS_ID}", file=sys.stderr) sys.exit(1) print(f"Ingress ID: {stream.ingress_id}") print(f"Title: {stream.title!r}") print(f"Status: {stream.status}") if __name__ == "__main__": main() ``` ```ts title={"Typescript (REST)"} import { LatticeClient } from "@anduril-industries/lattice-sdk"; const latticeEndpoint = process.env.LATTICE_ENDPOINT; const environmentToken = process.env.ENVIRONMENT_TOKEN; // Remove sandboxesToken from the following statements if you are not developing on Sandboxes. const sandboxesToken = process.env.SANDBOXES_TOKEN; if (!latticeEndpoint || !environmentToken || !sandboxesToken) { console.log('Missing required environment variables.'); process.exit(1); } const client = new LatticeClient( { baseUrl: `https://${latticeEndpoint}`, token: environmentToken, // Remove the following if you are not developing on Sandboxes. headers: { "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}` } } ); // Ingress ID of the stream to get. Replace before running. const ingressId = ""; async function App() { try { const response = await client.video.getIngressStream({ ingressId }); const stream = response.ingressStream; if (!stream) { console.log(`No ingress stream found for ${ingressId}`); return; } console.log(`Ingress ID: ${stream.ingressId}`); console.log(`Title: ${JSON.stringify(stream.title)}`); console.log(`Status: ${stream.status}`); } catch (error) { console.log(`Encountered the following error while fetching the stream: ${error}`); } } App(); ``` ```go title={"Go (gRPC)"} // This Go example is compatible with artifacts generated using // the following grpc/go plugin: https://buf.build/anduril/lattice-sdk/sdks/main:grpc/go package main import ( "context" "fmt" "log" "os" "time" "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/videomanager/v1/videomanagerv1grpc" videomanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/videomanager/v1" "google.golang.org/grpc" "google.golang.org/grpc/credentials" ) // Ingress ID of the stream to get. Replace before running. const ingressID = "" func main() { environmentToken := os.Getenv("ENVIRONMENT_TOKEN") latticeEndpoint := os.Getenv("LATTICE_ENDPOINT") sandboxesToken := os.Getenv("SANDBOXES_TOKEN") if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" { log.Fatalf("Missing required environment variables") } auth := &BearerTokenAuth{ Token: environmentToken, SandboxesToken: sandboxesToken, } opts := []grpc.DialOption{ grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")), grpc.WithPerRPCCredentials(auth), } conn, err := grpc.NewClient(latticeEndpoint, opts...) if err != nil { log.Fatalf("Did not connect: %v", err) } defer conn.Close() client := videomanagerv1grpc.NewVideoManagerAPIClient(conn) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() resp, err := client.GetIngressStream(ctx, &videomanagerv1.GetIngressStreamRequest{ IngressId: ingressID, }) if err != nil { log.Fatalf("Failed to get stream: %v", err) } stream := resp.GetIngressStream() fmt.Printf("Ingress ID: %s\n", stream.GetIngressId()) fmt.Printf("Title: %q\n", stream.GetTitle()) fmt.Printf("Status: %s\n", stream.GetStatus()) } ``` ```py title={"Python (gRPC)"} # This Python example is compatible with artifacts generated using the following # Buf plugins: https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:bufbuild/py # and https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:connectrpc/py import os import sys from bearer_auth import create_client from anduril.videomanager.v1.get_ingress_stream_request_pub_pb import ( GetIngressStreamRequest, ) from anduril.videomanager.v1.video_manager_api_pub_connect import ( VideoManagerAPIClientSync, ) # Ingress ID of the stream to get. Replace before running. INGRESS_ID = "" def main(): environment_token = os.getenv("ENVIRONMENT_TOKEN") lattice_endpoint = os.getenv("LATTICE_ENDPOINT") sandboxes_token = os.getenv("SANDBOXES_TOKEN") if not environment_token or not lattice_endpoint or not sandboxes_token: print("Missing required environment variables", file=sys.stderr) sys.exit(1) # Create the API client over Connect gRPC with authentication client = create_client(VideoManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token) try: response = client.get_ingress_stream( GetIngressStreamRequest(ingress_id=INGRESS_ID), timeout_ms=10000, ) except Exception as error: print(f"Failed to get stream: {error}", file=sys.stderr) sys.exit(1) if not response.has_field("ingress_stream"): print(f"No ingress stream found for {INGRESS_ID}", file=sys.stderr) sys.exit(1) stream = response.ingress_stream print(f"Ingress ID: {stream.ingress_id}") print(f"Title: {stream.title!r}") print(f"Status: {stream.status}") if __name__ == "__main__": main() ``` ```rs title={"Rust (gRPC)"} // This Rust example is compatible with artifacts generated using // the following grpc/rust plugin: https://buf.build/anduril/lattice-sdk/sdks/main:community/neoeinstein-tonic mod bearer_auth; use anduril_lattice_sdk_community_neoeinstein_prost::anduril::videomanager::v1::GetIngressStreamRequest; use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::videomanager::v1::tonic::video_manager_api_client::VideoManagerApiClient; use bearer_auth::EnvironmentTokenAuth; use tonic::metadata::MetadataValue; use tonic::transport::{Channel, ClientTlsConfig}; use tonic::Request; use std::env; // Ingress ID of the stream to get. Replace before running. const INGRESS_ID: &str = ""; #[tokio::main] async fn main() -> Result<(), Box> { let environment_token = env::var("ENVIRONMENT_TOKEN") .expect("ENVIRONMENT_TOKEN environment variable not set"); let lattice_endpoint = env::var("LATTICE_ENDPOINT") .expect("LATTICE_ENDPOINT environment variable not set"); let sandboxes_token = env::var("SANDBOXES_TOKEN") .expect("SANDBOXES_TOKEN environment variable not set"); let auth = EnvironmentTokenAuth::new( environment_token, sandboxes_token, ); let tls_config = ClientTlsConfig::new().with_native_roots(); let channel = Channel::from_shared(format!("https://{}", lattice_endpoint))? .tls_config(tls_config)? .connect() .await?; let access_token = auth.get_token().await?; let sandboxes_token = auth.sandboxes_token(); let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?; let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?; let mut client = VideoManagerApiClient::with_interceptor( channel, move |mut req: Request<()>| { req.metadata_mut() .insert("authorization", auth_header.clone()); req.metadata_mut() .insert("anduril-sandbox-authorization", sandbox_header.clone()); Ok(req) }, ); let request = Request::new(GetIngressStreamRequest { ingress_id: INGRESS_ID.to_string(), }); let response = client.get_ingress_stream(request).await?.into_inner(); match response.ingress_stream { Some(stream) => { println!("Ingress ID: {}", stream.ingress_id); println!("Title: {:?}", stream.title); println!("Status: {:?}", stream.status()); } None => eprintln!("No ingress stream found for {}", INGRESS_ID), } Ok(()) } ``` ## 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. ```go title={"Go (REST)"} // 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 = "" // Entity the video is attached to. Set to "" to only stop the stream without // editing any entity's media list. entityID = "" ) 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) } ``` ```java title={"Java (REST)"} package org.example; import java.time.OffsetDateTime; import java.time.ZoneOffset; import java.util.ArrayList; import java.util.List; import com.anduril.Lattice; import com.anduril.resources.entities.requests.EntityOverride; import com.anduril.types.Entity; import com.anduril.types.Media; import com.anduril.types.MediaItem; import com.anduril.types.Provenance; /** * Stops (archives) an ingress stream with the Lattice SDK and removes its stale media reference * from the associated camera entity. */ public class DeleteIngressStream { // Ingress ID of the stream to stop. Replace before running. private static final String INGRESS_ID = ""; // Entity the video is attached to. Set to "" to only stop the stream without // editing any entity's media list. private static final String ENTITY_ID = ""; public static void main(String[] args) { // Get environment variables String endpoint = System.getenv("LATTICE_ENDPOINT"); String environmentToken = System.getenv("ENVIRONMENT_TOKEN"); String sandboxesToken = System.getenv("SANDBOXES_TOKEN"); // Check required variables and ensure URL has https:// if (endpoint == null || environmentToken == null || sandboxesToken == null) { System.err.println("Missing required environment variables"); System.exit(1); } if (!endpoint.startsWith("https://")) endpoint = "https://" + endpoint; try { // Create Lattice client with Sandbox authentication Lattice client = Lattice.builder() .url(endpoint) .token(environmentToken) .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken) .build(); System.out.println("Stopping ingress stream with ID: " + INGRESS_ID); client.video().deleteIngressStream(INGRESS_ID); // With no entity to update, stopping the stream is all there is to do. if (ENTITY_ID.isEmpty()) { return; } // Read the entity, filter the stale MediaItem out, and write the surviving items back. Entity entity = client.entities().getEntity(ENTITY_ID); List current = entity.getMedia().flatMap(Media::getMedia).orElse(List.of()); List remaining = new ArrayList<>(); for (MediaItem item : current) { if (!item.getItemIdentifier().orElse("").equals(INGRESS_ID)) { remaining.add(item); } } if (remaining.size() == current.size()) { System.out.println("Video " + INGRESS_ID + " was not in entity " + ENTITY_ID + " media list; nothing to remove"); return; } EntityOverride entityOverride = EntityOverride.builder() .entity(Entity.builder() .entityId(ENTITY_ID) .media(Media.builder() .media(remaining) .build()) .build()) .provenance(Provenance.builder() .integrationName("lattice-video-integration") .dataType("video-association") .sourceUpdateTime(OffsetDateTime.now(ZoneOffset.UTC)) .build()) .build(); client.entities().overrideEntity(ENTITY_ID, "media", entityOverride); System.out.println("Removed video " + INGRESS_ID + " from entity " + ENTITY_ID); } catch (Exception e) { System.err.println("Exception: " + e.getMessage()); System.exit(1); } } } ``` ```py title={"Python (REST)"} # This Python example uses the Lattice REST SDK (anduril-lattice-sdk) to stop (archive) an ingress # stream and remove its stale media reference from the associated camera entity. import os import sys from datetime import datetime, timezone from anduril import Entity, Lattice, Media, Provenance # Ingress ID of the stream to stop. Replace before running. INGRESS_ID = "" # Entity the video is attached to. Set to "" to only stop the stream without # editing any entity's media list. ENTITY_ID = "" def main(): environment_token = os.getenv("ENVIRONMENT_TOKEN") lattice_endpoint = os.getenv("LATTICE_ENDPOINT") # Remove sandboxes_token from the following statements if you are not developing on Sandboxes. sandboxes_token = os.getenv("SANDBOXES_TOKEN") if not environment_token or not lattice_endpoint or not sandboxes_token: print("Missing required environment variables.", file=sys.stderr) sys.exit(1) client = Lattice( base_url=f"https://{lattice_endpoint}", token=lambda: str(environment_token), # Remove the following header if you are not developing on Sandboxes. headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"}, ) print(f"Stopping ingress stream with ID: {INGRESS_ID}") try: client.video.delete_ingress_stream(INGRESS_ID) except Exception as error: print(f"Failed to stop ingress stream: {error}", file=sys.stderr) sys.exit(1) # With no entity to update, stopping the stream is all there is to do. if not ENTITY_ID: return # Read the entity, filter the stale MediaItem out, and write the surviving items back. try: entity = client.entities.get_entity(ENTITY_ID) except Exception as error: print(f"Failed to fetch entity {ENTITY_ID}: {error}", file=sys.stderr) sys.exit(1) current = list(entity.media.media) if entity.media and entity.media.media else [] remaining = [item for item in current if item.item_identifier != INGRESS_ID] if len(remaining) == len(current): print( f"Video {INGRESS_ID} was not in entity {ENTITY_ID} media list; " "nothing to remove" ) return try: client.entities.override_entity( entity_id=ENTITY_ID, field_path="media", entity=Entity( entity_id=ENTITY_ID, media=Media(media=remaining), ), provenance=Provenance( integration_name="lattice-video-integration", data_type="video-association", source_update_time=datetime.now(timezone.utc), ), ) except Exception as error: print(f"Failed to override entity media: {error}", file=sys.stderr) sys.exit(1) print(f"Removed video {INGRESS_ID} from entity {ENTITY_ID}") if __name__ == "__main__": main() ``` ```ts title={"Typescript (REST)"} import { LatticeClient } from "@anduril-industries/lattice-sdk"; const latticeEndpoint = process.env.LATTICE_ENDPOINT; const environmentToken = process.env.ENVIRONMENT_TOKEN; // Remove sandboxesToken from the following statements if you are not developing on Sandboxes. const sandboxesToken = process.env.SANDBOXES_TOKEN; if (!latticeEndpoint || !environmentToken || !sandboxesToken) { console.log('Missing required environment variables.'); process.exit(1); } const client = new LatticeClient( { baseUrl: `https://${latticeEndpoint}`, token: environmentToken, // Remove the following if you are not developing on Sandboxes. headers: { "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}` } } ); // Ingress ID of the stream to stop. Replace before running. const ingressId = ""; // Entity the video is attached to. Set to "" to only stop the stream without // editing any entity's media list. const entityId = ""; async function App() { try { console.log(`Stopping ingress stream with ID: ${ingressId}`); await client.video.deleteIngressStream({ ingressId }); // 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. const entity = await client.entities.getEntity({ entityId }); const current = entity.media?.media ?? []; const remaining = current.filter((item) => item.itemIdentifier !== ingressId); if (remaining.length === current.length) { console.log(`Video ${ingressId} was not in entity ${entityId} media list; nothing to remove`); return; } // A field_path of media replaces the entire Media component with the surviving items. await client.entities.overrideEntity({ entityId, fieldPath: "media", entity: { entityId: entityId, media: { media: remaining } }, provenance: { integrationName: "lattice-video-integration", dataType: "video-association", sourceUpdateTime: new Date().toISOString() } }); console.log(`Removed video ${ingressId} from entity ${entityId}`); } catch (error) { console.log(`Encountered the following error while stopping the stream: ${error}`); } } App(); ``` ```go title={"Go (gRPC)"} // This Go example is compatible with artifacts generated using // the following grpc/go plugin: https://buf.build/anduril/lattice-sdk/sdks/main:grpc/go package main import ( "context" "fmt" "log" "os" "time" "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/videomanager/v1/videomanagerv1grpc" videomanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/videomanager/v1" "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/entitymanager/v1/entitymanagerv1grpc" entitymanager "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/entitymanager/v1" "google.golang.org/grpc" "google.golang.org/grpc/credentials" "google.golang.org/protobuf/types/known/timestamppb" ) const ( // Ingress ID of the stream to stop. Replace before running. ingressID = "" // Entity the video is attached to. Set to "" to only stop the stream without // editing any entity's media list. entityID = "" ) func main() { environmentToken := os.Getenv("ENVIRONMENT_TOKEN") latticeEndpoint := os.Getenv("LATTICE_ENDPOINT") sandboxesToken := os.Getenv("SANDBOXES_TOKEN") if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" { log.Fatalf("Missing required environment variables") } auth := &BearerTokenAuth{ Token: environmentToken, SandboxesToken: sandboxesToken, } opts := []grpc.DialOption{ grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")), grpc.WithPerRPCCredentials(auth), } conn, err := grpc.NewClient(latticeEndpoint, opts...) if err != nil { log.Fatalf("Did not connect: %v", err) } defer conn.Close() videoClient := videomanagerv1grpc.NewVideoManagerAPIClient(conn) emClient := entitymanagerv1grpc.NewEntityManagerAPIClient(conn) ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() fmt.Printf("Stopping ingress stream with ID: %s\n", ingressID) _, err = videoClient.DeleteIngressStream(ctx, &videomanagerv1.DeleteIngressStreamRequest{ IngressId: ingressID, }) if err != nil { log.Fatalf("Failed to stop ingress stream: %v", err) } // 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. getResp, err := emClient.GetEntity(ctx, &entitymanager.GetEntityRequest{EntityId: entityID}) if err != nil { log.Fatalf("Failed to fetch entity %s: %v", entityID, err) } current := getResp.GetEntity().GetMedia().GetMedia() remaining := make([]*entitymanager.MediaItem, 0, len(current)) for _, item := range current { if item.GetItemIdentifier() == ingressID { continue } 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 } _, err = emClient.OverrideEntity(ctx, &entitymanager.OverrideEntityRequest{ Entity: &entitymanager.Entity{ EntityId: entityID, Media: &entitymanager.Media{Media: remaining}, }, FieldPath: []string{"media"}, Provenance: &entitymanager.Provenance{ IntegrationName: "lattice-video-integration", DataType: "video-association", SourceUpdateTime: timestamppb.Now(), }, }) if err != nil { log.Fatalf("Failed to override entity media: %v", err) } fmt.Printf("Removed video %s from entity %s\n", ingressID, entityID) } ``` ```py title={"Python (gRPC)"} # This Python example is compatible with artifacts generated using the following # Buf plugins: https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:bufbuild/py # and https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:connectrpc/py import os import sys from datetime import datetime, timezone from bearer_auth import create_client from protobuf.wkt import Timestamp from anduril.entitymanager.v1.entity_pub_pb import Entity, Provenance from anduril.entitymanager.v1.entity_manager_api_pub_pb import ( GetEntityRequest, OverrideEntityRequest, ) from anduril.entitymanager.v1.entity_manager_api_pub_connect import EntityManagerAPIClientSync from anduril.entitymanager.v1.media_pub_pb import Media from anduril.videomanager.v1.delete_ingress_stream_request_pub_pb import ( DeleteIngressStreamRequest, ) from anduril.videomanager.v1.video_manager_api_pub_connect import ( VideoManagerAPIClientSync, ) # Ingress ID of the stream to stop. Replace before running. INGRESS_ID = "" # Entity the video is attached to. Set to "" to only stop the stream without # editing any entity's media list. ENTITY_ID = "" def main(): environment_token = os.getenv("ENVIRONMENT_TOKEN") lattice_endpoint = os.getenv("LATTICE_ENDPOINT") sandboxes_token = os.getenv("SANDBOXES_TOKEN") if not environment_token or not lattice_endpoint or not sandboxes_token: print("Missing required environment variables", file=sys.stderr) sys.exit(1) # Create the API clients over Connect gRPC with authentication video_client = create_client(VideoManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token) em_client = create_client(EntityManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token) print(f"Stopping ingress stream with ID: {INGRESS_ID}") try: video_client.delete_ingress_stream( DeleteIngressStreamRequest(ingress_id=INGRESS_ID), timeout_ms=30000, ) except Exception as error: print(f"Failed to stop ingress stream: {error}", file=sys.stderr) sys.exit(1) # With no entity to update, stopping the stream is all there is to do. if not ENTITY_ID: return # Read the entity, filter the stale MediaItem out, and write the surviving items back. try: get_resp = em_client.get_entity( GetEntityRequest(entity_id=ENTITY_ID), timeout_ms=30000, ) except Exception as error: print(f"Failed to fetch entity {ENTITY_ID}: {error}", file=sys.stderr) sys.exit(1) current = list(get_resp.entity.media.media) remaining = [item for item in current if item.item_identifier != INGRESS_ID] if len(remaining) == len(current): print( f"Video {INGRESS_ID} was not in entity {ENTITY_ID} media list; " "nothing to remove" ) return now_ts = Timestamp.from_datetime(datetime.now(timezone.utc)) try: em_client.override_entity( OverrideEntityRequest( entity=Entity( entity_id=ENTITY_ID, media=Media(media=remaining), ), field_path=["media"], provenance=Provenance( integration_name="lattice-video-integration", data_type="video-association", source_update_time=now_ts, ), ), timeout_ms=30000, ) except Exception as error: print(f"Failed to override entity media: {error}", file=sys.stderr) sys.exit(1) print(f"Removed video {INGRESS_ID} from entity {ENTITY_ID}") if __name__ == "__main__": main() ``` ```rs title={"Rust (gRPC)"} // This Rust example is compatible with artifacts generated using // the following grpc/rust plugin: https://buf.build/anduril/lattice-sdk/sdks/main:community/neoeinstein-tonic mod bearer_auth; use anduril_lattice_sdk_community_neoeinstein_prost::anduril::entitymanager::v1::{ Entity, GetEntityRequest, Media, OverrideEntityRequest, Provenance, }; use anduril_lattice_sdk_community_neoeinstein_prost::anduril::videomanager::v1::DeleteIngressStreamRequest; use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::entitymanager::v1::tonic::entity_manager_api_client::EntityManagerApiClient; use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::videomanager::v1::tonic::video_manager_api_client::VideoManagerApiClient; use bearer_auth::EnvironmentTokenAuth; use prost_types::Timestamp; use tonic::metadata::MetadataValue; use tonic::transport::{Channel, ClientTlsConfig}; use tonic::Request; use std::env; use std::time::{SystemTime, UNIX_EPOCH}; // Ingress ID of the stream to stop. Replace before running. const INGRESS_ID: &str = ""; // Entity the video is attached to. Set to "" to only stop the stream without // editing any entity's media list. const ENTITY_ID: &str = ""; fn now_timestamp() -> Timestamp { let d = SystemTime::now().duration_since(UNIX_EPOCH).unwrap(); Timestamp { seconds: d.as_secs() as i64, nanos: d.subsec_nanos() as i32, } } #[tokio::main] async fn main() -> Result<(), Box> { let environment_token = env::var("ENVIRONMENT_TOKEN") .expect("ENVIRONMENT_TOKEN environment variable not set"); let lattice_endpoint = env::var("LATTICE_ENDPOINT") .expect("LATTICE_ENDPOINT environment variable not set"); let sandboxes_token = env::var("SANDBOXES_TOKEN") .expect("SANDBOXES_TOKEN environment variable not set"); let auth = EnvironmentTokenAuth::new( environment_token, sandboxes_token, ); let tls_config = ClientTlsConfig::new().with_native_roots(); let channel = Channel::from_shared(format!("https://{}", lattice_endpoint))? .tls_config(tls_config)? .connect() .await?; let access_token = auth.get_token().await?; let sandboxes_token = auth.sandboxes_token(); let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?; let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?; // Both clients share the same channel and inject the same auth metadata. let video_auth = auth_header.clone(); let video_sandbox = sandbox_header.clone(); let mut video_client = VideoManagerApiClient::with_interceptor( channel.clone(), move |mut req: Request<()>| { req.metadata_mut().insert("authorization", video_auth.clone()); req.metadata_mut() .insert("anduril-sandbox-authorization", video_sandbox.clone()); Ok(req) }, ); let mut em_client = EntityManagerApiClient::with_interceptor( channel, move |mut req: Request<()>| { req.metadata_mut() .insert("authorization", auth_header.clone()); req.metadata_mut() .insert("anduril-sandbox-authorization", sandbox_header.clone()); Ok(req) }, ); println!("Stopping ingress stream with ID: {}", INGRESS_ID); video_client .delete_ingress_stream(Request::new(DeleteIngressStreamRequest { ingress_id: INGRESS_ID.to_string(), })) .await?; // With no entity to update, stopping the stream is all there is to do. if ENTITY_ID.is_empty() { return Ok(()); } // Read the entity, filter the stale MediaItem out, and write the surviving items back. let get_resp = em_client .get_entity(Request::new(GetEntityRequest { entity_id: ENTITY_ID.to_string(), })) .await? .into_inner(); let current = get_resp .entity .and_then(|e| e.media) .map(|m| m.media) .unwrap_or_default(); let original_len = current.len(); let remaining: Vec<_> = current .into_iter() .filter(|item| item.item_identifier != INGRESS_ID) .collect(); if remaining.len() == original_len { println!( "Video {} was not in entity {} media list; nothing to remove", INGRESS_ID, ENTITY_ID ); return Ok(()); } // A field_path of media replaces the entire Media component with the surviving items. let override_req = Request::new(OverrideEntityRequest { entity: Some(Entity { entity_id: ENTITY_ID.to_string(), media: Some(Media { media: remaining }), ..Default::default() }), field_path: vec!["media".to_string()], provenance: Some(Provenance { integration_name: "lattice-video-integration".to_string(), data_type: "video-association".to_string(), source_update_time: Some(now_timestamp()), ..Default::default() }), }); match em_client.override_entity(override_req).await { Ok(_) => println!("Removed video {} from entity {}", INGRESS_ID, ENTITY_ID), Err(e) => eprintln!("Failed to override entity media: {}", e), } Ok(()) } ``` ## 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. ```go title={"Go (REST)"} // 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() } } ``` ```java title={"Java (REST)"} package org.example; import java.util.List; import com.anduril.Lattice; import com.anduril.resources.video.requests.ListEgressStreamsRequest; import com.anduril.types.EgressStream; import com.anduril.types.ListEgressStreamsResponse; /** * Lists active egress streams with the Lattice SDK. */ public class ListEgressStreams { public static void main(String[] args) { // Get environment variables String endpoint = System.getenv("LATTICE_ENDPOINT"); String environmentToken = System.getenv("ENVIRONMENT_TOKEN"); String sandboxesToken = System.getenv("SANDBOXES_TOKEN"); // Check required variables and ensure URL has https:// if (endpoint == null || environmentToken == null || sandboxesToken == null) { System.err.println("Missing required environment variables"); System.exit(1); } if (!endpoint.startsWith("https://")) endpoint = "https://" + endpoint; try { // Create Lattice client with Sandbox authentication Lattice client = Lattice.builder() .url(endpoint) .token(environmentToken) .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken) .build(); // Iterate every page of active streams. Keep all request fields identical between // pages other than page_token; the server may reject a request that changes them. String pageToken = ""; while (true) { ListEgressStreamsResponse response = client.video().listEgressStreams( ListEgressStreamsRequest.builder() .pageToken(pageToken) .build()); List streams = response.getEgressStreams().orElse(List.of()); for (EgressStream stream : streams) { System.out.println("Egress " + stream.getEgressId().orElse("") + " re-publishes ingress " + stream.getIngressId().orElse("")); } String nextPageToken = response.getNextPageToken().orElse(""); if (nextPageToken.isEmpty()) { break; } pageToken = nextPageToken; } } catch (Exception e) { System.err.println("Failed to list egress streams: " + e.getMessage()); System.exit(1); } } } ``` ```py title={"Python (REST)"} # This Python example uses the Lattice REST SDK (anduril-lattice-sdk) to list active egress # streams. import os import sys from anduril import Lattice def main(): environment_token = os.getenv("ENVIRONMENT_TOKEN") lattice_endpoint = os.getenv("LATTICE_ENDPOINT") # Remove sandboxes_token from the following statements if you are not developing on Sandboxes. sandboxes_token = os.getenv("SANDBOXES_TOKEN") if not environment_token or not lattice_endpoint or not sandboxes_token: print("Missing required environment variables.", file=sys.stderr) sys.exit(1) client = Lattice( base_url=f"https://{lattice_endpoint}", token=lambda: str(environment_token), # Remove the following header if you are not developing on Sandboxes. headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"}, ) # Iterate every page of active streams. Keep all request fields identical between # pages other than page_token; the server may reject a request that changes them. page_token = "" while True: try: response = client.video.list_egress_streams(page_token=page_token) except Exception as error: print(f"Failed to list egress streams: {error}", file=sys.stderr) sys.exit(1) for stream in response.egress_streams or []: print(f"Egress {stream.egress_id} re-publishes ingress {stream.ingress_id}") if not response.next_page_token: break page_token = response.next_page_token if __name__ == "__main__": main() ``` ```ts title={"Typescript (REST)"} import { LatticeClient } from "@anduril-industries/lattice-sdk"; const latticeEndpoint = process.env.LATTICE_ENDPOINT; const environmentToken = process.env.ENVIRONMENT_TOKEN; // Remove sandboxesToken from the following statements if you are not developing on Sandboxes. const sandboxesToken = process.env.SANDBOXES_TOKEN; if (!latticeEndpoint || !environmentToken || !sandboxesToken) { console.log('Missing required environment variables.'); process.exit(1); } const client = new LatticeClient( { baseUrl: `https://${latticeEndpoint}`, token: environmentToken, // Remove the following if you are not developing on Sandboxes. headers: { "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}` } } ); async function App() { try { // 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. let pageToken = ""; while (true) { const response = await client.video.listEgressStreams({ pageToken }); for (const stream of response.egressStreams ?? []) { console.log(`Egress ${stream.egressId} re-publishes ingress ${stream.ingressId}`); } if (!response.nextPageToken) { break; } pageToken = response.nextPageToken; } } catch (error) { console.log(`Encountered the following error while listing streams: ${error}`); } } App(); ``` ```go title={"Go (gRPC)"} // This Go example is compatible with artifacts generated using // the following grpc/go plugin: https://buf.build/anduril/lattice-sdk/sdks/main:grpc/go package main import ( "context" "fmt" "log" "os" "time" "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/videomanager/v1/videomanagerv1grpc" videomanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/videomanager/v1" "google.golang.org/grpc" "google.golang.org/grpc/credentials" ) func main() { environmentToken := os.Getenv("ENVIRONMENT_TOKEN") latticeEndpoint := os.Getenv("LATTICE_ENDPOINT") sandboxesToken := os.Getenv("SANDBOXES_TOKEN") if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" { log.Fatalf("Missing required environment variables") } auth := &BearerTokenAuth{ Token: environmentToken, SandboxesToken: sandboxesToken, } opts := []grpc.DialOption{ grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")), grpc.WithPerRPCCredentials(auth), } conn, err := grpc.NewClient(latticeEndpoint, opts...) if err != nil { log.Fatalf("Did not connect: %v", err) } defer conn.Close() client := videomanagerv1grpc.NewVideoManagerAPIClient(conn) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() // Iterate every page of active streams. Keep all request fields identical between // pages other than page_token; the server may reject a request that changes them. pageToken := "" for { resp, err := client.ListEgressStreams(ctx, &videomanagerv1.ListEgressStreamsRequest{ PageToken: pageToken, }) if err != nil { log.Fatalf("Failed to list egress streams: %v", err) } for _, stream := range resp.GetEgressStreams() { fmt.Printf("Egress %s re-publishes ingress %s\n", stream.GetEgressId(), stream.GetIngressId()) } if resp.GetNextPageToken() == "" { break } pageToken = resp.GetNextPageToken() } } ``` ```py title={"Python (gRPC)"} # This Python example is compatible with artifacts generated using the following # Buf plugins: https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:bufbuild/py # and https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:connectrpc/py import os import sys from bearer_auth import create_client from anduril.videomanager.v1.list_egress_streams_request_pub_pb import ( ListEgressStreamsRequest, ) from anduril.videomanager.v1.video_manager_api_pub_connect import ( VideoManagerAPIClientSync, ) def main(): environment_token = os.getenv("ENVIRONMENT_TOKEN") lattice_endpoint = os.getenv("LATTICE_ENDPOINT") sandboxes_token = os.getenv("SANDBOXES_TOKEN") if not environment_token or not lattice_endpoint or not sandboxes_token: print("Missing required environment variables", file=sys.stderr) sys.exit(1) # Create the API client over Connect gRPC with authentication client = create_client(VideoManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token) # Iterate every page of active streams. Keep all request fields identical between # pages other than page_token; the server may reject a request that changes them. page_token = "" while True: try: response = client.list_egress_streams( ListEgressStreamsRequest(page_token=page_token), timeout_ms=10000, ) except Exception as error: print(f"Failed to list egress streams: {error}", file=sys.stderr) sys.exit(1) for stream in response.egress_streams: print(f"Egress {stream.egress_id} re-publishes ingress {stream.ingress_id}") if not response.next_page_token: break page_token = response.next_page_token if __name__ == "__main__": main() ``` ```rs title={"Rust (gRPC)"} // This Rust example is compatible with artifacts generated using // the following grpc/rust plugin: https://buf.build/anduril/lattice-sdk/sdks/main:community/neoeinstein-tonic mod bearer_auth; use anduril_lattice_sdk_community_neoeinstein_prost::anduril::videomanager::v1::ListEgressStreamsRequest; use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::videomanager::v1::tonic::video_manager_api_client::VideoManagerApiClient; use bearer_auth::EnvironmentTokenAuth; use tonic::metadata::MetadataValue; use tonic::transport::{Channel, ClientTlsConfig}; use tonic::Request; use std::env; #[tokio::main] async fn main() -> Result<(), Box> { let environment_token = env::var("ENVIRONMENT_TOKEN") .expect("ENVIRONMENT_TOKEN environment variable not set"); let lattice_endpoint = env::var("LATTICE_ENDPOINT") .expect("LATTICE_ENDPOINT environment variable not set"); let sandboxes_token = env::var("SANDBOXES_TOKEN") .expect("SANDBOXES_TOKEN environment variable not set"); let auth = EnvironmentTokenAuth::new( environment_token, sandboxes_token, ); let tls_config = ClientTlsConfig::new().with_native_roots(); let channel = Channel::from_shared(format!("https://{}", lattice_endpoint))? .tls_config(tls_config)? .connect() .await?; let access_token = auth.get_token().await?; let sandboxes_token = auth.sandboxes_token(); let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?; let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?; let mut client = VideoManagerApiClient::with_interceptor( channel, move |mut req: Request<()>| { req.metadata_mut() .insert("authorization", auth_header.clone()); req.metadata_mut() .insert("anduril-sandbox-authorization", sandbox_header.clone()); Ok(req) }, ); // Iterate every page of active streams. Keep all request fields identical between // pages other than page_token; the server may reject a request that changes them. let mut page_token = String::new(); loop { let request = Request::new(ListEgressStreamsRequest { page_size: 0, page_token: page_token.clone(), }); let response = client.list_egress_streams(request).await?.into_inner(); for stream in &response.egress_streams { println!( "Egress {} re-publishes ingress {}", stream.egress_id, stream.ingress_id ); } if response.next_page_token.is_empty() { break; } page_token = response.next_page_token; } Ok(()) } ``` ## 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. ```go title={"Go (REST)"} // 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 = "" 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()) } } ``` ```java title={"Java (REST)"} package org.example; import com.anduril.Lattice; import com.anduril.types.EgressStream; import com.anduril.types.GetEgressStreamResponse; import com.anduril.types.RtspEgress; import com.anduril.types.SrtEgress; /** * Retrieves a single egress stream with the Lattice SDK. */ public class GetEgressStream { // Egress ID of the stream to get. Replace before running. private static final String EGRESS_ID = ""; public static void main(String[] args) { // Get environment variables String endpoint = System.getenv("LATTICE_ENDPOINT"); String environmentToken = System.getenv("ENVIRONMENT_TOKEN"); String sandboxesToken = System.getenv("SANDBOXES_TOKEN"); // Check required variables and ensure URL has https:// if (endpoint == null || environmentToken == null || sandboxesToken == null) { System.err.println("Missing required environment variables"); System.exit(1); } if (!endpoint.startsWith("https://")) endpoint = "https://" + endpoint; try { // Create Lattice client with Sandbox authentication Lattice client = Lattice.builder() .url(endpoint) .token(environmentToken) .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken) .build(); GetEgressStreamResponse response = client.video().getEgressStream(EGRESS_ID); EgressStream stream = response.getEgressStream().orElse(null); if (stream == null) { System.err.println("No egress stream found for " + EGRESS_ID); System.exit(1); } System.out.println("Egress ID: " + stream.getEgressId().orElse("")); System.out.println("Ingress ID: " + stream.getIngressId().orElse("")); // Exactly one of rtsp or srt is set, matching the transport the egress stream uses. RtspEgress rtsp = stream.getRtsp().orElse(null); SrtEgress srt = stream.getSrt().orElse(null); if (rtsp != null) { System.out.println("RTSP URL: " + rtsp.getUrl().orElse("")); } else if (srt != null) { System.out.println("SRT URL: " + srt.getUrl().orElse("")); System.out.println("Session ID: " + srt.getSessionId().orElse("")); } } catch (Exception e) { System.err.println("Failed to get egress stream: " + e.getMessage()); System.exit(1); } } } ``` ```py title={"Python (REST)"} # This Python example uses the Lattice REST SDK (anduril-lattice-sdk) to retrieve a single egress # stream. import os import sys from anduril import Lattice # Egress ID of the stream to get. Replace before running. EGRESS_ID = "" def main(): environment_token = os.getenv("ENVIRONMENT_TOKEN") lattice_endpoint = os.getenv("LATTICE_ENDPOINT") # Remove sandboxes_token from the following statements if you are not developing on Sandboxes. sandboxes_token = os.getenv("SANDBOXES_TOKEN") if not environment_token or not lattice_endpoint or not sandboxes_token: print("Missing required environment variables.", file=sys.stderr) sys.exit(1) client = Lattice( base_url=f"https://{lattice_endpoint}", token=lambda: str(environment_token), # Remove the following header if you are not developing on Sandboxes. headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"}, ) try: response = client.video.get_egress_stream(EGRESS_ID) except Exception as error: print(f"Failed to get egress stream: {error}", file=sys.stderr) sys.exit(1) stream = response.egress_stream if not stream: print(f"No egress stream found for {EGRESS_ID}", file=sys.stderr) sys.exit(1) print(f"Egress ID: {stream.egress_id}") print(f"Ingress ID: {stream.ingress_id}") # Exactly one of rtsp or srt is set, matching the transport the egress stream uses. if stream.rtsp: print(f"RTSP URL: {stream.rtsp.url}") elif stream.srt: print(f"SRT URL: {stream.srt.url}") print(f"Session ID: {stream.srt.session_id}") if __name__ == "__main__": main() ``` ```ts title={"Typescript (REST)"} import { LatticeClient } from "@anduril-industries/lattice-sdk"; const latticeEndpoint = process.env.LATTICE_ENDPOINT; const environmentToken = process.env.ENVIRONMENT_TOKEN; // Remove sandboxesToken from the following statements if you are not developing on Sandboxes. const sandboxesToken = process.env.SANDBOXES_TOKEN; if (!latticeEndpoint || !environmentToken || !sandboxesToken) { console.log('Missing required environment variables.'); process.exit(1); } const client = new LatticeClient( { baseUrl: `https://${latticeEndpoint}`, token: environmentToken, // Remove the following if you are not developing on Sandboxes. headers: { "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}` } } ); // Egress ID of the stream to get. Replace before running. const egressId = ""; async function App() { try { const response = await client.video.getEgressStream({ egressId }); const stream = response.egressStream; if (!stream) { console.log(`No egress stream found for ${egressId}`); return; } console.log(`Egress ID: ${stream.egressId}`); console.log(`Ingress ID: ${stream.ingressId}`); // Exactly one of rtsp or srt is set, matching the transport the egress stream uses. if (stream.rtsp) { console.log(`RTSP URL: ${stream.rtsp.url}`); } else if (stream.srt) { console.log(`SRT URL: ${stream.srt.url}`); console.log(`Session ID: ${stream.srt.sessionId}`); } } catch (error) { console.log(`Encountered the following error while fetching the stream: ${error}`); } } App(); ``` ```go title={"Go (gRPC)"} // This Go example is compatible with artifacts generated using // the following grpc/go plugin: https://buf.build/anduril/lattice-sdk/sdks/main:grpc/go package main import ( "context" "fmt" "log" "os" "time" "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/videomanager/v1/videomanagerv1grpc" videomanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/videomanager/v1" "google.golang.org/grpc" "google.golang.org/grpc/credentials" ) // Egress ID of the stream to get. Replace before running. const egressID = "" func main() { environmentToken := os.Getenv("ENVIRONMENT_TOKEN") latticeEndpoint := os.Getenv("LATTICE_ENDPOINT") sandboxesToken := os.Getenv("SANDBOXES_TOKEN") if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" { log.Fatalf("Missing required environment variables") } auth := &BearerTokenAuth{ Token: environmentToken, SandboxesToken: sandboxesToken, } opts := []grpc.DialOption{ grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")), grpc.WithPerRPCCredentials(auth), } conn, err := grpc.NewClient(latticeEndpoint, opts...) if err != nil { log.Fatalf("Did not connect: %v", err) } defer conn.Close() client := videomanagerv1grpc.NewVideoManagerAPIClient(conn) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() resp, err := client.GetEgressStream(ctx, &videomanagerv1.GetEgressStreamRequest{ EgressId: egressID, }) if err != nil { log.Fatalf("Failed to get egress stream: %v", err) } stream := resp.GetEgressStream() 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 rtsp := stream.GetRtsp(); rtsp != nil { fmt.Printf("RTSP URL: %s\n", rtsp.GetUrl()) } else if srt := stream.GetSrt(); srt != nil { fmt.Printf("SRT URL: %s\n", srt.GetUrl()) fmt.Printf("Session ID: %s\n", srt.GetSessionId()) } } ``` ```py title={"Python (gRPC)"} # This Python example is compatible with artifacts generated using the following # Buf plugins: https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:bufbuild/py # and https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:connectrpc/py import os import sys from bearer_auth import create_client from anduril.videomanager.v1.get_egress_stream_request_pub_pb import ( GetEgressStreamRequest, ) from anduril.videomanager.v1.video_manager_api_pub_connect import ( VideoManagerAPIClientSync, ) # Egress ID of the stream to get. Replace before running. EGRESS_ID = "" def main(): environment_token = os.getenv("ENVIRONMENT_TOKEN") lattice_endpoint = os.getenv("LATTICE_ENDPOINT") sandboxes_token = os.getenv("SANDBOXES_TOKEN") if not environment_token or not lattice_endpoint or not sandboxes_token: print("Missing required environment variables", file=sys.stderr) sys.exit(1) # Create the API client over Connect gRPC with authentication client = create_client(VideoManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token) try: response = client.get_egress_stream( GetEgressStreamRequest(egress_id=EGRESS_ID), timeout_ms=10000, ) except Exception as error: print(f"Failed to get egress stream: {error}", file=sys.stderr) sys.exit(1) if not response.has_field("egress_stream"): print(f"No egress stream found for {EGRESS_ID}", file=sys.stderr) sys.exit(1) stream = response.egress_stream print(f"Egress ID: {stream.egress_id}") print(f"Ingress ID: {stream.ingress_id}") # Exactly one of rtsp or srt is set, matching the transport the egress stream uses. if stream.has_field("rtsp"): print(f"RTSP URL: {stream.egress.value.url}") elif stream.has_field("srt"): print(f"SRT URL: {stream.egress.value.url}") print(f"Session ID: {stream.egress.value.session_id}") if __name__ == "__main__": main() ``` ```rs title={"Rust (gRPC)"} // This Rust example is compatible with artifacts generated using // the following grpc/rust plugin: https://buf.build/anduril/lattice-sdk/sdks/main:community/neoeinstein-tonic mod bearer_auth; use anduril_lattice_sdk_community_neoeinstein_prost::anduril::videomanager::v1::{ egress_stream::Egress, GetEgressStreamRequest, }; use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::videomanager::v1::tonic::video_manager_api_client::VideoManagerApiClient; use bearer_auth::EnvironmentTokenAuth; use tonic::metadata::MetadataValue; use tonic::transport::{Channel, ClientTlsConfig}; use tonic::Request; use std::env; // Egress ID of the stream to get. Replace before running. const EGRESS_ID: &str = ""; #[tokio::main] async fn main() -> Result<(), Box> { let environment_token = env::var("ENVIRONMENT_TOKEN") .expect("ENVIRONMENT_TOKEN environment variable not set"); let lattice_endpoint = env::var("LATTICE_ENDPOINT") .expect("LATTICE_ENDPOINT environment variable not set"); let sandboxes_token = env::var("SANDBOXES_TOKEN") .expect("SANDBOXES_TOKEN environment variable not set"); let auth = EnvironmentTokenAuth::new( environment_token, sandboxes_token, ); let tls_config = ClientTlsConfig::new().with_native_roots(); let channel = Channel::from_shared(format!("https://{}", lattice_endpoint))? .tls_config(tls_config)? .connect() .await?; let access_token = auth.get_token().await?; let sandboxes_token = auth.sandboxes_token(); let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?; let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?; let mut client = VideoManagerApiClient::with_interceptor( channel, move |mut req: Request<()>| { req.metadata_mut() .insert("authorization", auth_header.clone()); req.metadata_mut() .insert("anduril-sandbox-authorization", sandbox_header.clone()); Ok(req) }, ); let request = Request::new(GetEgressStreamRequest { egress_id: EGRESS_ID.to_string(), }); let response = client.get_egress_stream(request).await?.into_inner(); match response.egress_stream { Some(stream) => { println!("Egress ID: {}", stream.egress_id); println!("Ingress ID: {}", stream.ingress_id); // Exactly one of rtsp or srt is set, matching the transport the egress stream uses. match stream.egress { Some(Egress::Rtsp(rtsp)) => println!("RTSP URL: {}", rtsp.url), Some(Egress::Srt(srt)) => { println!("SRT URL: {}", srt.url); println!("Session ID: {}", srt.session_id); } None => {} } } None => eprintln!("No egress stream found for {}", EGRESS_ID), } Ok(()) } ``` ## 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. ```go title={"Go (REST)"} // 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 = "" 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) } ``` ```java title={"Java (REST)"} package org.example; import com.anduril.Lattice; /** * Stops an egress stream with the Lattice SDK, which halts delivery to its downstream consumers. */ public class DeleteEgressStream { // Egress ID of the stream to stop. Replace before running. private static final String EGRESS_ID = ""; public static void main(String[] args) { // Get environment variables String endpoint = System.getenv("LATTICE_ENDPOINT"); String environmentToken = System.getenv("ENVIRONMENT_TOKEN"); String sandboxesToken = System.getenv("SANDBOXES_TOKEN"); // Check required variables and ensure URL has https:// if (endpoint == null || environmentToken == null || sandboxesToken == null) { System.err.println("Missing required environment variables"); System.exit(1); } if (!endpoint.startsWith("https://")) endpoint = "https://" + endpoint; try { // Create Lattice client with Sandbox authentication Lattice client = Lattice.builder() .url(endpoint) .token(environmentToken) .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken) .build(); // Deleting an egress stream stops only that downstream delivery. The underlying ingress // stream and any other egress streams consuming it are unaffected, and no entity or // media reference is touched. Deleting an egress stream that does not exist returns // NOT_FOUND. System.out.println("Stopping egress stream with ID: " + EGRESS_ID); client.video().deleteEgressStream(EGRESS_ID); System.out.println("Stopped egress stream " + EGRESS_ID); } catch (Exception e) { System.err.println("Failed to stop egress stream: " + e.getMessage()); System.exit(1); } } } ``` ```py title={"Python (REST)"} # This Python example uses the Lattice REST SDK (anduril-lattice-sdk) to stop an egress stream, # which halts delivery to its downstream consumers. import os import sys from anduril import Lattice # Egress ID of the stream to stop. Replace before running. EGRESS_ID = "" def main(): environment_token = os.getenv("ENVIRONMENT_TOKEN") lattice_endpoint = os.getenv("LATTICE_ENDPOINT") # Remove sandboxes_token from the following statements if you are not developing on Sandboxes. sandboxes_token = os.getenv("SANDBOXES_TOKEN") if not environment_token or not lattice_endpoint or not sandboxes_token: print("Missing required environment variables.", file=sys.stderr) sys.exit(1) client = Lattice( base_url=f"https://{lattice_endpoint}", token=lambda: str(environment_token), # Remove the following header if you are not developing on Sandboxes. headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"}, ) # 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. print(f"Stopping egress stream with ID: {EGRESS_ID}") try: client.video.delete_egress_stream(EGRESS_ID) except Exception as error: print(f"Failed to stop egress stream: {error}", file=sys.stderr) sys.exit(1) print(f"Stopped egress stream {EGRESS_ID}") if __name__ == "__main__": main() ``` ```ts title={"Typescript (REST)"} import { LatticeClient } from "@anduril-industries/lattice-sdk"; const latticeEndpoint = process.env.LATTICE_ENDPOINT; const environmentToken = process.env.ENVIRONMENT_TOKEN; // Remove sandboxesToken from the following statements if you are not developing on Sandboxes. const sandboxesToken = process.env.SANDBOXES_TOKEN; if (!latticeEndpoint || !environmentToken || !sandboxesToken) { console.log('Missing required environment variables.'); process.exit(1); } const client = new LatticeClient( { baseUrl: `https://${latticeEndpoint}`, token: environmentToken, // Remove the following if you are not developing on Sandboxes. headers: { "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}` } } ); // Egress ID of the stream to stop. Replace before running. const egressId = ""; async function App() { try { // Deleting an egress stream stops only that downstream delivery. The underlying ingress // stream and any other egress streams consuming it are unaffected, and no entity's media // list is touched. Deleting an egress stream that does not exist returns NOT_FOUND. console.log(`Stopping egress stream with ID: ${egressId}`); await client.video.deleteEgressStream({ egressId }); console.log(`Stopped egress stream ${egressId}`); } catch (error) { console.log(`Encountered the following error while stopping the stream: ${error}`); } } App(); ``` ```go title={"Go (gRPC)"} // This Go example is compatible with artifacts generated using // the following grpc/go plugin: https://buf.build/anduril/lattice-sdk/sdks/main:grpc/go package main import ( "context" "fmt" "log" "os" "time" "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/videomanager/v1/videomanagerv1grpc" videomanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/videomanager/v1" "google.golang.org/grpc" "google.golang.org/grpc/credentials" ) // Egress ID of the stream to stop. Replace before running. const egressID = "" func main() { environmentToken := os.Getenv("ENVIRONMENT_TOKEN") latticeEndpoint := os.Getenv("LATTICE_ENDPOINT") sandboxesToken := os.Getenv("SANDBOXES_TOKEN") if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" { log.Fatalf("Missing required environment variables") } auth := &BearerTokenAuth{ Token: environmentToken, SandboxesToken: sandboxesToken, } opts := []grpc.DialOption{ grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")), grpc.WithPerRPCCredentials(auth), } conn, err := grpc.NewClient(latticeEndpoint, opts...) if err != nil { log.Fatalf("Did not connect: %v", err) } defer conn.Close() client := videomanagerv1grpc.NewVideoManagerAPIClient(conn) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() // Deleting an egress stream stops only that downstream delivery. The underlying // ingress stream and any other egress streams consuming it are unaffected, and no // entity or media is touched. Deleting an egress stream that does not exist returns // NOT_FOUND. fmt.Printf("Stopping egress stream with ID: %s\n", egressID) _, err = client.DeleteEgressStream(ctx, &videomanagerv1.DeleteEgressStreamRequest{ EgressId: egressID, }) if err != nil { log.Fatalf("Failed to stop egress stream: %v", err) } fmt.Printf("Stopped egress stream %s\n", egressID) } ``` ```py title={"Python (gRPC)"} # This Python example is compatible with artifacts generated using the following # Buf plugins: https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:bufbuild/py # and https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:connectrpc/py import os import sys from bearer_auth import create_client from anduril.videomanager.v1.delete_egress_stream_request_pub_pb import ( DeleteEgressStreamRequest, ) from anduril.videomanager.v1.video_manager_api_pub_connect import ( VideoManagerAPIClientSync, ) # Egress ID of the stream to stop. Replace before running. EGRESS_ID = "" def main(): environment_token = os.getenv("ENVIRONMENT_TOKEN") lattice_endpoint = os.getenv("LATTICE_ENDPOINT") sandboxes_token = os.getenv("SANDBOXES_TOKEN") if not environment_token or not lattice_endpoint or not sandboxes_token: print("Missing required environment variables", file=sys.stderr) sys.exit(1) # Create the API client over Connect gRPC with authentication client = create_client(VideoManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token) # Deleting an egress stream stops only that downstream delivery. The underlying ingress # stream and any other egress streams consuming it are unaffected, and no entity or # media is touched. Deleting an egress stream that does not exist returns NOT_FOUND. print(f"Stopping egress stream with ID: {EGRESS_ID}") try: client.delete_egress_stream( DeleteEgressStreamRequest(egress_id=EGRESS_ID), timeout_ms=30000, ) except Exception as error: print(f"Failed to stop egress stream: {error}", file=sys.stderr) sys.exit(1) print(f"Stopped egress stream {EGRESS_ID}") if __name__ == "__main__": main() ``` ```rs title={"Rust (gRPC)"} // This Rust example is compatible with artifacts generated using // the following grpc/rust plugin: https://buf.build/anduril/lattice-sdk/sdks/main:community/neoeinstein-tonic mod bearer_auth; use anduril_lattice_sdk_community_neoeinstein_prost::anduril::videomanager::v1::DeleteEgressStreamRequest; use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::videomanager::v1::tonic::video_manager_api_client::VideoManagerApiClient; use bearer_auth::EnvironmentTokenAuth; use tonic::metadata::MetadataValue; use tonic::transport::{Channel, ClientTlsConfig}; use tonic::Request; use std::env; // Egress ID of the stream to stop. Replace before running. const EGRESS_ID: &str = ""; #[tokio::main] async fn main() -> Result<(), Box> { let environment_token = env::var("ENVIRONMENT_TOKEN") .expect("ENVIRONMENT_TOKEN environment variable not set"); let lattice_endpoint = env::var("LATTICE_ENDPOINT") .expect("LATTICE_ENDPOINT environment variable not set"); let sandboxes_token = env::var("SANDBOXES_TOKEN") .expect("SANDBOXES_TOKEN environment variable not set"); let auth = EnvironmentTokenAuth::new( environment_token, sandboxes_token, ); let tls_config = ClientTlsConfig::new().with_native_roots(); let channel = Channel::from_shared(format!("https://{}", lattice_endpoint))? .tls_config(tls_config)? .connect() .await?; let access_token = auth.get_token().await?; let sandboxes_token = auth.sandboxes_token(); let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?; let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?; let mut client = VideoManagerApiClient::with_interceptor( channel, move |mut req: Request<()>| { req.metadata_mut() .insert("authorization", auth_header.clone()); req.metadata_mut() .insert("anduril-sandbox-authorization", sandbox_header.clone()); Ok(req) }, ); // Deleting an egress stream stops only that downstream delivery. Unlike deleting an // ingress stream, it never touches an entity or its media; the underlying ingress // stream and any other egress streams consuming it are unaffected. Deleting an egress // stream that does not exist returns NOT_FOUND. println!("Stopping egress stream with ID: {}", EGRESS_ID); client .delete_egress_stream(Request::new(DeleteEgressStreamRequest { egress_id: EGRESS_ID.to_string(), })) .await?; println!("Stopped egress stream {}", EGRESS_ID); Ok(()) } ``` ## What's next * [Overview](/guides/video/overview): Review the API surface and stream lifecycle. * [Create streams](/guides/video/create-streams): Publish an asset and start an ingress stream. > List, retrieve, stop, and re-egress existing video streams.