From c8cadde384beee83cd4ded7247728c285926792a Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Wed, 4 Mar 2026 15:56:16 -0800 Subject: [PATCH] live-dashboard: allow customization of ingest fields --- .../components/live-dashboard/stream-key.tsx | 65 ++++++++------- js/components/src/streamplace-store/index.tsx | 1 + .../src/streamplace-store/ingest.tsx | 32 ++++++++ .../streamplace-store/streamplace-store.tsx | 13 ++- .../ingest/place-stream-ingest-defs.md | 52 ++++++++++++ .../place-stream-ingest-getingesturls.md | 67 ++++++++++++++++ .../content/docs/lex-reference/openapi.json | 48 +++++++++++ lexicons/place/stream/ingest/defs.json | 22 ++++++ .../place/stream/ingest/getIngestUrls.json | 31 ++++++++ pkg/atproto/sync.go | 1 - pkg/config/config.go | 13 +++ pkg/spxrpc/place_stream_ingest.go | 79 +++++++++++++++++++ pkg/spxrpc/stubs.go | 14 ++++ pkg/statedb/queue_processor.go | 1 - pkg/streamplace/ingestdefs.go | 16 ++++ pkg/streamplace/ingestgetIngestUrls.go | 57 +++++++++++++ 16 files changed, 479 insertions(+), 33 deletions(-) create mode 100644 js/components/src/streamplace-store/ingest.tsx create mode 100644 js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-defs.md create mode 100644 js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-getingesturls.md create mode 100644 lexicons/place/stream/ingest/defs.json create mode 100644 lexicons/place/stream/ingest/getIngestUrls.json create mode 100644 pkg/spxrpc/place_stream_ingest.go create mode 100644 pkg/streamplace/ingestdefs.go create mode 100644 pkg/streamplace/ingestgetIngestUrls.go diff --git a/js/app/components/live-dashboard/stream-key.tsx b/js/app/components/live-dashboard/stream-key.tsx index 9ee00a4b..0a35447b 100644 --- a/js/app/components/live-dashboard/stream-key.tsx +++ b/js/app/components/live-dashboard/stream-key.tsx @@ -5,17 +5,20 @@ import { Code, Row, Text, + useStreamplaceStore, useTheme, useToast, View, zero, } from "@streamplace/components"; +import useGetIngests from "@streamplace/components/src/streamplace-store/ingest"; import Loading from "components/loading/loading"; import { Clipboard, ClipboardCheck } from "lucide-react-native"; import { useEffect, useState } from "react"; import { ScrollView, TextInput } from "react-native"; import { useStore } from "store"; import { useIsReady, useUserProfile } from "store/hooks"; +import { PlaceStreamIngestDefs } from "streamplace"; const FormRow = ({ children }: { children: React.ReactNode }) => { return ( @@ -42,12 +45,19 @@ const Content = ({ children }: { children: React.ReactNode }) => { }; export function StreamKeyScreen() { - const [protocol, setProtocol] = useState<"whip" | "rtmp">("rtmp"); + const [ingest, setIngest] = useState( + null, + ); const isReady = useIsReady(); const userProfile = useUserProfile(); const openLoginModal = useStore((state) => state.openLoginModal); const route = useRoute(); const url = useStore((state) => state.url); + const ingests = useStreamplaceStore((state) => state.ingests); + const getIngests = useGetIngests(); + useEffect(() => { + getIngests(); + }, []); useEffect(() => { if (isReady && !userProfile) { @@ -55,6 +65,12 @@ export function StreamKeyScreen() { } }, [isReady, userProfile, openLoginModal, route.name, route.params]); + useEffect(() => { + if (ingests && ingests.length > 0 && !ingest) { + setIngest(ingests[0]); + } + }, [ingests, ingest]); + if (!isReady) { return ; } @@ -63,36 +79,30 @@ export function StreamKeyScreen() { return ; } + if (!ingests) { + return ; + } + return ( - - + {ingests.map((ing, i) => ( + + ))} - {protocol === "whip" && } - {protocol === "rtmp" && } + {ingest?.type === "whip" && } + {ingest?.type.startsWith("rtmp") && ( + + )} @@ -168,9 +178,6 @@ export function WHIPDescription({ url }: { url: string }) { } export function RTMPDescription({ url }: { url: string }) { - const u = new URL(url); - const rtmpUrl = `rtmps://${u.host}:1935/live`; - return ( <> @@ -183,7 +190,7 @@ export function RTMPDescription({ url }: { url: string }) { state.setIngests); + + return async () => { + if (!pdsAgent || !did) { + throw new Error("No PDS agent or DID available"); + } + + const result = await pdsAgent.place.stream.ingest.getIngestUrls(); + if (!result.success) { + throw new Error("Failed to get ingests"); + } + + const ingests = result.data.ingests + .map((ingest) => { + if (PlaceStreamIngestDefs.isIngest(ingest)) { + return ingest; + } + console.error("Invalid ingest", ingest); + return null; + }) + .filter((ingest) => ingest !== null); + + setIngests(ingests); + }; +} diff --git a/js/components/src/streamplace-store/streamplace-store.tsx b/js/components/src/streamplace-store/streamplace-store.tsx index 513ce322..2d7662a3 100644 --- a/js/components/src/streamplace-store/streamplace-store.tsx +++ b/js/components/src/streamplace-store/streamplace-store.tsx @@ -1,6 +1,10 @@ import { SessionManager } from "@atproto/api/dist/session-manager"; import { useContext } from "react"; -import { PlaceStreamChatProfile, PlaceStreamLivestream } from "streamplace"; +import { + PlaceStreamChatProfile, + PlaceStreamIngestDefs, + PlaceStreamLivestream, +} from "streamplace"; import { createStore, StoreApi, useStore } from "zustand"; import storage from "../storage"; import { StreamplaceContext } from "../streamplace-provider/context"; @@ -41,6 +45,9 @@ export interface StreamplaceState { handle: string | null; chatProfile: PlaceStreamChatProfile.Record | null; + ingests: PlaceStreamIngestDefs.Ingest[] | null; + setIngests: (ingests: PlaceStreamIngestDefs.Ingest[] | null) => void; + // Content metadata state contentMetadata: ContentMetadataResult | null; setContentMetadata: (metadata: ContentMetadataResult | null) => void; @@ -113,7 +120,9 @@ export const makeStreamplaceStore = ({ oauthSession: null, handle: null, chatProfile: null, - + ingests: null, + setIngests: (ingests: PlaceStreamIngestDefs.Ingest[] | null) => + set({ ingests: ingests }), broadcasterDID: null, setBroadcasterDID: (broadcasterDID: string | null) => set({ broadcasterDID }), diff --git a/js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-defs.md b/js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-defs.md new file mode 100644 index 00000000..50f85802 --- /dev/null +++ b/js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-defs.md @@ -0,0 +1,52 @@ +--- +title: place.stream.ingest.defs +description: Reference for the place.stream.ingest.defs lexicon +--- + +**Lexicon Version:** 1 + +## Definitions + + + +### `ingest` + +**Type:** `object` + +An ingest URL for a Streamplace station. + +**Properties:** + +| Name | Type | Req'd | Description | Constraints | +| ------ | -------- | ----- | ----------------------------------------------------------------------- | ------------- | +| `type` | `string` | ✅ | The type of ingest endpoint, currently 'rtmp' and 'whip' are supported. | | +| `url` | `string` | ✅ | The URL of the ingest endpoint. | Format: `uri` | + +--- + +## Lexicon Source + +```json +{ + "lexicon": 1, + "id": "place.stream.ingest.defs", + "defs": { + "ingest": { + "type": "object", + "description": "An ingest URL for a Streamplace station.", + "required": ["type", "url"], + "properties": { + "type": { + "type": "string", + "description": "The type of ingest endpoint, currently 'rtmp' and 'whip' are supported." + }, + "url": { + "type": "string", + "format": "uri", + "description": "The URL of the ingest endpoint." + } + } + } + } +} +``` diff --git a/js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-getingesturls.md b/js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-getingesturls.md new file mode 100644 index 00000000..a7d44f67 --- /dev/null +++ b/js/docs/src/content/docs/lex-reference/ingest/place-stream-ingest-getingesturls.md @@ -0,0 +1,67 @@ +--- +title: place.stream.ingest.getIngestUrls +description: Reference for the place.stream.ingest.getIngestUrls lexicon +--- + +**Lexicon Version:** 1 + +## Definitions + + + +### `main` + +**Type:** `query` + +Get ingest URLs for a Streamplace station. + +**Parameters:** _(None defined)_ + +**Output:** + +- **Encoding:** `application/json` +- **Schema:** + +**Schema Type:** `object` + +| Name | Type | Req'd | Description | Constraints | +| --------- | ---------------------------------------------------------------------------------------------------------------------- | ----- | ----------- | ----------- | +| `ingests` | Array of Union of:
  [`place.stream.ingest.defs#ingest`](/lex-reference/place-stream-ingest-defs#ingest) | ✅ | | | + +--- + +## Lexicon Source + +```json +{ + "lexicon": 1, + "id": "place.stream.ingest.getIngestUrls", + "defs": { + "main": { + "type": "query", + "description": "Get ingest URLs for a Streamplace station.", + "parameters": { + "type": "params", + "properties": {} + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["ingests"], + "properties": { + "ingests": { + "type": "array", + "items": { + "type": "union", + "refs": ["place.stream.ingest.defs#ingest"] + } + } + } + } + }, + "errors": [] + } + } +} +``` diff --git a/js/docs/src/content/docs/lex-reference/openapi.json b/js/docs/src/content/docs/lex-reference/openapi.json index af71e48f..dbcdff4a 100644 --- a/js/docs/src/content/docs/lex-reference/openapi.json +++ b/js/docs/src/content/docs/lex-reference/openapi.json @@ -1683,6 +1683,38 @@ } } }, + "/xrpc/place.stream.ingest.getIngestUrls": { + "get": { + "summary": "Get ingest URLs for a Streamplace station.", + "operationId": "place.stream.ingest.getIngestUrls", + "tags": ["place.stream.ingest"], + "responses": { + "200": { + "description": "Success", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "ingests": { + "type": "array", + "items": { + "oneOf": [ + { + "$ref": "#/components/schemas/place.stream.ingest.defs_ingest" + } + ] + } + } + }, + "required": ["ingests"] + } + } + } + } + } + } + }, "/xrpc/place.stream.graph.getFollowingUser": { "get": { "summary": "Get whether or not user A is following user B.", @@ -3600,6 +3632,22 @@ "format": "byte", "description": "MP4 file of a user's signed livestream segment" }, + "place.stream.ingest.defs_ingest": { + "type": "object", + "description": "An ingest URL for a Streamplace station.", + "properties": { + "type": { + "type": "string", + "description": "The type of ingest endpoint, currently 'rtmp' and 'whip' are supported." + }, + "url": { + "type": "string", + "description": "The URL of the ingest endpoint.", + "format": "uri" + } + }, + "required": ["type", "url"] + }, "com.atproto.repo.strongRef": { "type": "object", "properties": { diff --git a/lexicons/place/stream/ingest/defs.json b/lexicons/place/stream/ingest/defs.json new file mode 100644 index 00000000..d206a078 --- /dev/null +++ b/lexicons/place/stream/ingest/defs.json @@ -0,0 +1,22 @@ +{ + "lexicon": 1, + "id": "place.stream.ingest.defs", + "defs": { + "ingest": { + "type": "object", + "description": "An ingest URL for a Streamplace station.", + "required": ["type", "url"], + "properties": { + "type": { + "type": "string", + "description": "The type of ingest endpoint, currently 'rtmp' and 'whip' are supported." + }, + "url": { + "type": "string", + "format": "uri", + "description": "The URL of the ingest endpoint." + } + } + } + } +} diff --git a/lexicons/place/stream/ingest/getIngestUrls.json b/lexicons/place/stream/ingest/getIngestUrls.json new file mode 100644 index 00000000..6fb5844d --- /dev/null +++ b/lexicons/place/stream/ingest/getIngestUrls.json @@ -0,0 +1,31 @@ +{ + "lexicon": 1, + "id": "place.stream.ingest.getIngestUrls", + "defs": { + "main": { + "type": "query", + "description": "Get ingest URLs for a Streamplace station.", + "parameters": { + "type": "params", + "properties": {} + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["ingests"], + "properties": { + "ingests": { + "type": "array", + "items": { + "type": "union", + "refs": ["place.stream.ingest.defs#ingest"] + } + } + } + } + }, + "errors": [] + } + } +} diff --git a/pkg/atproto/sync.go b/pkg/atproto/sync.go index 5483bf3c..13610a55 100644 --- a/pkg/atproto/sync.go +++ b/pkg/atproto/sync.go @@ -392,7 +392,6 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD // if we check after exactly rec.IdleTimeoutSeconds we might miss the finalization by a few seconds scheduledAt = scheduledAt.Add((time.Duration(*rec.IdleTimeoutSeconds) * time.Second) + (10 * time.Second)).UTC() taskKey := fmt.Sprintf("finalize-livestream::%s::%s", aturi.String(), scheduledAt.Format(util.ISO8601)) - log.Warn(ctx, "queueing stream finalization task", "taskKey", taskKey, "scheduledAt", scheduledAt) _, err = atsync.StatefulDB.EnqueueTask(ctx, statedb.TaskFinalizeLivestream, task, statedb.WithTaskKey(taskKey), statedb.WithScheduledAt(scheduledAt)) if err != nil { return fmt.Errorf("failed to enqueue remove red circle task: %w", err) diff --git a/pkg/config/config.go b/pkg/config/config.go index 3df0d7ce..63e4708f 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -31,6 +31,7 @@ import ( "stream.place/streamplace/pkg/crypto/aqpub" "stream.place/streamplace/pkg/integrations/discord/discordtypes" "stream.place/streamplace/pkg/log" + placestream "stream.place/streamplace/pkg/streamplace" ) const SPDataDir = "$SP_DATA_DIR" @@ -146,6 +147,7 @@ type CLI struct { AdminDIDs []string Syndicate []string PlayerTelemetry bool + Ingests *placestream.IngestGetIngestUrls_Output } // ContentFilters represents the content filtering configuration @@ -814,6 +816,17 @@ func (cli *CLI) NewCommand(name string) *urfavecli.Command { Destination: &cli.LocalDBURL, Sources: urfavecli.EnvVars("SP_LOCAL_DB_URL"), }, + &urfavecli.StringFlag{ + Name: "ingests", + Usage: `JSON array of ingests to return from place.stream.ingest.getIngestUrls. Default is auto-generated ingests for RTMP and WHIP`, + Action: func(ctx context.Context, cmd *urfavecli.Command, s string) error { + if s == "" { + return nil + } + return json.Unmarshal([]byte(s), &cli.Ingests) + }, + Sources: urfavecli.EnvVars("SP_INGESTS"), + }, &urfavecli.BoolFlag{ Name: "external-signing", Usage: "DEPRECATED, does nothing.", diff --git a/pkg/spxrpc/place_stream_ingest.go b/pkg/spxrpc/place_stream_ingest.go new file mode 100644 index 00000000..6275ed4e --- /dev/null +++ b/pkg/spxrpc/place_stream_ingest.go @@ -0,0 +1,79 @@ +package spxrpc + +import ( + "context" + "fmt" + "net" + "strings" + + placestream "stream.place/streamplace/pkg/streamplace" +) + +func hostPort(host, port, defaultPort string) string { + if port == defaultPort { + return host + } + return net.JoinHostPort(host, port) +} + +func (s *Server) handlePlaceStreamIngestGetIngestUrls(ctx context.Context) (*placestream.IngestGetIngestUrls_Output, error) { + if s.cli.Ingests != nil { + return s.cli.Ingests, nil + } + broadcasterDID := s.cli.BroadcasterDID() + broadcasterHost := strings.TrimPrefix(broadcasterDID, "did:web:") + out := &placestream.IngestGetIngestUrls_Output{ + Ingests: []*placestream.IngestGetIngestUrls_Output_Ingests_Elem{}, + } + if !s.cli.Secure { + _, rtmpPort, err := net.SplitHostPort(s.cli.RTMPAddr) + if err != nil { + return nil, err + } + rtmpUrl := fmt.Sprintf("rtmp://%s/live", hostPort(broadcasterHost, rtmpPort, "1935")) + out.Ingests = append(out.Ingests, &placestream.IngestGetIngestUrls_Output_Ingests_Elem{ + IngestDefs_Ingest: &placestream.IngestDefs_Ingest{ + Type: "rtmp", + Url: rtmpUrl, + }, + }) + + _, httpPort, err := net.SplitHostPort(s.cli.HTTPAddr) + if err != nil { + return nil, err + } + whipUrl := fmt.Sprintf("http://%s", hostPort(broadcasterHost, httpPort, "80")) + out.Ingests = append(out.Ingests, &placestream.IngestGetIngestUrls_Output_Ingests_Elem{ + IngestDefs_Ingest: &placestream.IngestDefs_Ingest{ + Type: "whip", + Url: whipUrl, + }, + }) + } else { + _, rtmpsPort, err := net.SplitHostPort(s.cli.RTMPSAddr) + if err != nil { + return nil, err + } + rtmpsUrl := fmt.Sprintf("rtmps://%s:%s/live", broadcasterHost, rtmpsPort) + out.Ingests = append(out.Ingests, &placestream.IngestGetIngestUrls_Output_Ingests_Elem{ + IngestDefs_Ingest: &placestream.IngestDefs_Ingest{ + Type: "rtmps", + Url: rtmpsUrl, + }, + }) + + _, httpsPort, err := net.SplitHostPort(s.cli.HTTPSAddr) + if err != nil { + return nil, err + } + whipUrl := fmt.Sprintf("https://%s", hostPort(broadcasterHost, httpsPort, "443")) + out.Ingests = append(out.Ingests, &placestream.IngestGetIngestUrls_Output_Ingests_Elem{ + IngestDefs_Ingest: &placestream.IngestDefs_Ingest{ + Type: "whip", + Url: whipUrl, + }, + }) + } + + return out, nil +} diff --git a/pkg/spxrpc/stubs.go b/pkg/spxrpc/stubs.go index ea388dca..7367ebc7 100644 --- a/pkg/spxrpc/stubs.go +++ b/pkg/spxrpc/stubs.go @@ -285,6 +285,7 @@ func (s *Server) RegisterHandlersPlaceStream(e *echo.Echo) error { e.POST("/xrpc/place.stream.branding.updateBlob", s.HandlePlaceStreamBrandingUpdateBlob) e.GET("/xrpc/place.stream.broadcast.getBroadcaster", s.HandlePlaceStreamBroadcastGetBroadcaster) e.GET("/xrpc/place.stream.graph.getFollowingUser", s.HandlePlaceStreamGraphGetFollowingUser) + e.GET("/xrpc/place.stream.ingest.getIngestUrls", s.HandlePlaceStreamIngestGetIngestUrls) e.POST("/xrpc/place.stream.live.denyTeleport", s.HandlePlaceStreamLiveDenyTeleport) e.GET("/xrpc/place.stream.live.getLiveUsers", s.HandlePlaceStreamLiveGetLiveUsers) e.GET("/xrpc/place.stream.live.getProfileCard", s.HandlePlaceStreamLiveGetProfileCard) @@ -419,6 +420,19 @@ func (s *Server) HandlePlaceStreamGraphGetFollowingUser(c echo.Context) error { return c.JSON(200, out) } +func (s *Server) HandlePlaceStreamIngestGetIngestUrls(c echo.Context) error { + ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamIngestGetIngestUrls") + defer span.End() + var out *placestream.IngestGetIngestUrls_Output + var handleErr error + // func (s *Server) handlePlaceStreamIngestGetIngestUrls(ctx context.Context) (*placestream.IngestGetIngestUrls_Output, error) + out, handleErr = s.handlePlaceStreamIngestGetIngestUrls(ctx) + if handleErr != nil { + return handleErr + } + return c.JSON(200, out) +} + func (s *Server) HandlePlaceStreamLiveDenyTeleport(c echo.Context) error { ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamLiveDenyTeleport") defer span.End() diff --git a/pkg/statedb/queue_processor.go b/pkg/statedb/queue_processor.go index 64a6fa48..a21983b0 100644 --- a/pkg/statedb/queue_processor.go +++ b/pkg/statedb/queue_processor.go @@ -80,7 +80,6 @@ func (state *StatefulDB) processTask(ctx context.Context, task *AppTask) error { func (state *StatefulDB) processFinalizeLivestreamTask(ctx context.Context, task *AppTask) error { ctx = log.WithLogValues(ctx, "func", "processFinalizeLivestreamTask") log.Debug(ctx, "processing finalize livestream task") - log.Warn(ctx, "processing finalize livestream task") var finalizeLivestreamTask FinalizeLivestreamTask if err := json.Unmarshal(task.Payload, &finalizeLivestreamTask); err != nil { return err diff --git a/pkg/streamplace/ingestdefs.go b/pkg/streamplace/ingestdefs.go new file mode 100644 index 00000000..8492dbae --- /dev/null +++ b/pkg/streamplace/ingestdefs.go @@ -0,0 +1,16 @@ +// Code generated by cmd/lexgen (see Makefile's lexgen); DO NOT EDIT. + +// Lexicon schema: place.stream.ingest.defs + +package streamplace + +// IngestDefs_Ingest is a "ingest" in the place.stream.ingest.defs schema. +// +// An ingest URL for a Streamplace station. +type IngestDefs_Ingest struct { + LexiconTypeID string `json:"$type" cborgen:"$type,const=place.stream.ingest.defs#ingest"` + // type: The type of ingest endpoint, currently 'rtmp' and 'whip' are supported. + Type string `json:"type" cborgen:"type"` + // url: The URL of the ingest endpoint. + Url string `json:"url" cborgen:"url"` +} diff --git a/pkg/streamplace/ingestgetIngestUrls.go b/pkg/streamplace/ingestgetIngestUrls.go new file mode 100644 index 00000000..5b6ee6d1 --- /dev/null +++ b/pkg/streamplace/ingestgetIngestUrls.go @@ -0,0 +1,57 @@ +// Code generated by cmd/lexgen (see Makefile's lexgen); DO NOT EDIT. + +// Lexicon schema: place.stream.ingest.getIngestUrls + +package streamplace + +import ( + "context" + "encoding/json" + "fmt" + + lexutil "github.com/bluesky-social/indigo/lex/util" +) + +// IngestGetIngestUrls_Output is the output of a place.stream.ingest.getIngestUrls call. +type IngestGetIngestUrls_Output struct { + Ingests []*IngestGetIngestUrls_Output_Ingests_Elem `json:"ingests" cborgen:"ingests"` +} + +type IngestGetIngestUrls_Output_Ingests_Elem struct { + IngestDefs_Ingest *IngestDefs_Ingest +} + +func (t *IngestGetIngestUrls_Output_Ingests_Elem) MarshalJSON() ([]byte, error) { + if t.IngestDefs_Ingest != nil { + t.IngestDefs_Ingest.LexiconTypeID = "place.stream.ingest.defs#ingest" + return json.Marshal(t.IngestDefs_Ingest) + } + return nil, fmt.Errorf("can not marshal empty union as JSON") +} + +func (t *IngestGetIngestUrls_Output_Ingests_Elem) UnmarshalJSON(b []byte) error { + typ, err := lexutil.TypeExtract(b) + if err != nil { + return err + } + + switch typ { + case "place.stream.ingest.defs#ingest": + t.IngestDefs_Ingest = new(IngestDefs_Ingest) + return json.Unmarshal(b, t.IngestDefs_Ingest) + default: + return nil + } +} + +// IngestGetIngestUrls calls the XRPC method "place.stream.ingest.getIngestUrls". +func IngestGetIngestUrls(ctx context.Context, c lexutil.LexClient) (*IngestGetIngestUrls_Output, error) { + var out IngestGetIngestUrls_Output + + params := map[string]interface{}{} + if err := c.LexDo(ctx, lexutil.Query, "", "place.stream.ingest.getIngestUrls", params, nil, &out); err != nil { + return nil, err + } + + return &out, nil +} -- 2.51.2