Skip to main content

Webhook-based stream state management

Native Frame's webhook system allows you to implement custom, server-side logic for managing your streams in real-time. By listening to and reacting to the Program States webhook event, you can dynamically control stream access, duration, and viewer permissions based on your own business rules. This guide will walk you through an example use case of setting up time-limited streams so that broadcasters and viewers can only stream for a limited amount of time.

Prerequisites​

  1. Node.js (version 18.17.0 or higher) installed on your system
  2. npm (Node Package Manager) installed
  3. A modern web browser that supports WebRTC (such as Chrome, Firefox, or Safari)
  4. Basic knowledge of JavaScript and web development
  5. A Native Frame account with access to the platform dashboard

You will also need to set up your webhook and make sure you understand webhook authentication and how to set up a public URL for your webhook server.

Step 1: Setting Up the Node.js Server​

First, we'll set up a basic Node.js server using Express.

Initialize Your Project​

Create a new directory for your project and initialize it with npm:

mkdir my-webhook-server
cd my-webhook-server
npm init -y

Install Dependencies​

We'll use Express for the server and cors to parse incoming webhook requests:

npm install express cors

Setting Up the Server and Implementing the Webhook Endpoint​

Begin by creating an index.js file in the root of your project:

touch index.js

In index.js, start by importing the Express framework:

const express = require('express');

Initialize the Express application:

const app = express();

Set up middleware to handle Cross-Origin Resource Sharing (CORS) and parse incoming JSON requests. This is essential for handling webhook requests coming from the Native Frame domain and ensuring the server can understand the JSON payload:

app.use(cors());
app.use(express.json());

Next, implement the /webhook POST endpoint that will receive webhook events from Native Frame. The request body carries programs and tokens at its top level (there is no wrapper object). Extract both — they contain the data about programs, streams, and users that need to be processed:

app.post('/webhook', (req, res) => {
  const { programs = {}, tokens = [] } = req.body;

  // Process the webhook data and prepare a response
  const response = {
    tokens: [],
    programs: {}
  };

  // We'll fill this in the next steps

  res.json(response);
});
Don't ship this stub as-is

The response must mirror the request: every program, stream, and token you receive has to be echoed back with verdict fields set. Anything omitted is treated as stop: true — the empty response above would stop every stream in the request, including the broadcaster. The next steps build the full echo.

Finally, start the server by having it listen on port 3000 or any port specified in the environment variables. This will make your server ready to accept incoming webhook requests:

const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
  console.log(`Server running on port ${PORT}`);
});

In this step, we've set up a basic Express server that listens for POST requests on the /webhook endpoint. The server is configured to parse incoming JSON data and handle CORS, which is crucial for receiving and processing webhook events from Native Frame. The /webhook endpoint extracts the programs object from the request body, which we'll use to process the webhook data and send an appropriate response back to Native Frame.

Step 2: Handling Webhook Requests​

We'll implement the logic to manage stream and viewer states based on the webhook events. You can learn more about the webhook request and response structure here.

Initialize In-Memory Storage​

At the top of your index.js, add two maps for the stream and viewer states:

const streamStates = new Map();
const viewerStates = new Map();

const MAX_STREAM_DURATION = 60000; // 1 minute for demo
const MAX_VIEWER_DURATION = 10000;  // 10 seconds for demo

These maps will keep track of the state of the stream and all viewers on the stream. This demo implementation uses in-memory storage, but you can use any storage mechanism that suits your needs.

Implementing State Management Logic​

Now we'll enhance the /webhook endpoint to include logic that manages the state of streams and viewers. This involves initializing streams and viewers, tracking their durations, and constructing a response that instructs Native Frame on how to handle each stream and viewer.

Update your webhook endpoint in index.js as follows:

app.post('/webhook', async (req, res) => {
  // The body is the program-states payload itself — no wrapper object
  const { programs = {}, tokens = [] } = req.body;

  // Get the current timestamp
  const now = Date.now();

  // Initialize the response object — it must mirror the request
  const response = { tokens: [], programs: {} };

  // Echo top-level tokens (users not yet associated with a program)
  for (const token of tokens) {
    response.tokens.push({ value: token.value, event: token.event, stop: false });
  }

  // Iterate over each program in the request
  for (const [programId, program] of Object.entries(programs)) {
    const programResponse = {
      // By default, do not stop the program
      stop: false,
      // Require authentication so viewer tokens are sent to this webhook
      needAuth: true,
      // Echo program-level tokens
      tokens: (program.tokens || []).map((token) => ({
        value: token.value,
        event: token.event,
        stop: false
      })),
      // Initialize the streams object
      streams: {}
    };
    response.programs[programId] = programResponse;

    // Iterate over each stream within the program
    for (const [streamId, stream] of Object.entries(program.streams || {})) {
      // If the stream state doesn't exist, initialize it
      if (!streamStates.has(streamId)) {
        streamStates.set(streamId, {
          // Record the stream's start time
          startTime: now,
          // The stream is initially not stopped
          stopped: false
        });

        // Schedule the stream to stop after MAX_STREAM_DURATION
        setTimeout(() => stopStream(streamId), MAX_STREAM_DURATION);
      }

      // Get the current state of the stream
      const streamState = streamStates.get(streamId);
      const streamStopReason = streamState.stopped
        ? 'Stream duration limit reached'
        : undefined;

      // Build the response for this stream
      const streamResponse = {
        // Require authentication for the stream
        needAuth: true,
        // Indicate whether the stream should be stopped
        stop: streamState.stopped,
        // Provide a stop reason if stopped
        stopReason: streamStopReason,
        producing: [],
        consuming: []
      };
      programResponse.streams[streamId] = streamResponse;

      // Echo every producing (broadcaster) entry.
      // For RTMP broadcasts, producing.token.value is the stream's authKey.
      for (const producing of stream.producing || []) {
        streamResponse.producing.push({
          streamName: producing.streamName,
          token: {
            value: producing.token.value,
            event: producing.token.event,
            stop: streamState.stopped,
            stopReason: streamStopReason,
            // The user id and name would come from your authorization system
            appData: {
              'user.scope': 'broadcaster',
              'user.id': 'broadcaster123',
              'user.name': 'John Doe'
            }
          }
        });
      }

      // Echo every consuming (viewer) entry
      for (const consuming of stream.consuming || []) {
        // Use the token value as the viewer ID
        const viewerId = consuming.token.value;

        // If the viewer state doesn't exist, initialize it
        if (!viewerStates.has(viewerId)) {
          viewerStates.set(viewerId, {
            // Record the viewer's start time
            startTime: now,
            // The viewer is initially not stopped
            stopped: false
          });

          // Schedule the viewer to be stopped after MAX_VIEWER_DURATION
          setTimeout(() => stopViewer(viewerId), MAX_VIEWER_DURATION);
        }

        // Get the current state of the viewer
        const viewerState = viewerStates.get(viewerId);

        streamResponse.consuming.push({
          streamName: consuming.streamName,
          token: {
            value: viewerId,
            event: consuming.token.event,
            stop: viewerState.stopped,
            stopReason: viewerState.stopped ? 'Viewer time limit reached' : undefined,
            appData: {
              'user.scope': 'viewer',
              'user.id': 'viewer456',
              'user.name': 'Jane Smith'
            }
          }
        });
      }
    }
  }

  // Send the response back to Native Frame
  res.json(response);
});

In this code, we're performing several key operations:

  • Stream Management: For each stream, we check if it's a new stream and initialize its state if necessary. We record the start time and set up a timer (setTimeout) to stop the stream after a predefined maximum duration (MAX_STREAM_DURATION). This simulates a scenario where streams have a limited lifespan.

  • Viewer Management: Similarly, for each viewer attempting to join a stream (each entry in a stream's consuming array), we initialize their state if they are new. We record their start time and schedule them to be stopped after MAX_VIEWER_DURATION. This controls how long a viewer can watch a stream.

  • Response Construction: We build a response that fully mirrors the request — every program, stream, producing entry, consuming entry, and token is echoed back with its verdict fields (stop, needAuth, appData). This is required, not optional: any entry missing from the response is treated as stop: true by Native Frame, so a partial response would tear down streams and viewers you never intended to touch.

This implementation allows us to dynamically control streams and viewers based on custom logic. By tracking start times and enforcing duration limits, we can simulate real-world scenarios where streams and viewer sessions have time constraints.

Step 3: Managing Stream and Viewer States​

Next, we'll add the functions to stop the stream and viewer, which is as simple as updating the stored state, which will get sent along with the webhook response the next time it's triggered.

Add State Management Functions​

At the bottom of your index.js, add:

function stopStream(streamId) {
  const streamState = streamStates.get(streamId);
  if (streamState) {
    streamState.stopped = true;
    console.log(`Stream ${streamId} stopped after ${MAX_STREAM_DURATION / 1000} seconds`);
  }
}

function stopViewer(viewerId) {
  const viewerState = viewerStates.get(viewerId);
  if (viewerState) {
    viewerState.stopped = true;
    console.log(`Viewer ${viewerId} stopped after ${MAX_VIEWER_DURATION / 1000} seconds`);
  }
}

The stored state flows into the stop fields of the mirrored response the next time the webhook fires — and when Native Frame receives stop: true for a stream or token, it stops that stream or viewer.

Final Code Snippet​

Below is the complete index.js file for your Node.js server:

// index.js

const express = require('express');
const cors = require('cors');

const app = express();

// Middleware to handle CORS and parse JSON requests
app.use(cors());
app.use(express.json());

// In-memory storage for stream and viewer states
const streamStates = new Map();
const viewerStates = new Map();

// Maximum durations (in milliseconds)
const MAX_STREAM_DURATION = 60000; // 1 minute for demo
const MAX_VIEWER_DURATION = 10000;  // 10 seconds for demo

function stopStream(streamId) {
  const streamState = streamStates.get(streamId);
  if (streamState) {
    streamState.stopped = true;
    console.log(`Stream ${streamId} stopped after ${MAX_STREAM_DURATION / 1000} seconds`);
  }
}

function stopViewer(viewerId) {
  const viewerState = viewerStates.get(viewerId);
  if (viewerState) {
    viewerState.stopped = true;
    console.log(`Viewer ${viewerId} stopped after ${MAX_VIEWER_DURATION / 1000} seconds`);
  }
}

app.post('/webhook', async (req, res) => {
  // The body is the program-states payload itself — no wrapper object
  const { programs = {}, tokens = [] } = req.body;

  // Get the current timestamp
  const now = Date.now();

  // Initialize the response object — it must mirror the request
  const response = { tokens: [], programs: {} };

  // Echo top-level tokens (users not yet associated with a program)
  for (const token of tokens) {
    response.tokens.push({ value: token.value, event: token.event, stop: false });
  }

  // Iterate over each program in the request
  for (const [programId, program] of Object.entries(programs)) {
    const programResponse = {
      // By default, do not stop the program
      stop: false,
      // Require authentication so viewer tokens are sent to this webhook
      needAuth: true,
      // Echo program-level tokens
      tokens: (program.tokens || []).map((token) => ({
        value: token.value,
        event: token.event,
        stop: false
      })),
      // Initialize the streams object
      streams: {}
    };
    response.programs[programId] = programResponse;

    // Iterate over each stream within the program
    for (const [streamId, stream] of Object.entries(program.streams || {})) {
      // If the stream state doesn't exist, initialize it
      if (!streamStates.has(streamId)) {
        streamStates.set(streamId, {
          // Record the stream's start time
          startTime: now,
          // The stream is initially not stopped
          stopped: false
        });

        // Schedule the stream to stop after MAX_STREAM_DURATION
        setTimeout(() => stopStream(streamId), MAX_STREAM_DURATION);
      }

      // Get the current state of the stream
      const streamState = streamStates.get(streamId);
      const streamStopReason = streamState.stopped
        ? 'Stream duration limit reached'
        : undefined;

      // Build the response for this stream
      const streamResponse = {
        // Require authentication for the stream
        needAuth: true,
        // Indicate whether the stream should be stopped
        stop: streamState.stopped,
        // Provide a stop reason if stopped
        stopReason: streamStopReason,
        producing: [],
        consuming: []
      };
      programResponse.streams[streamId] = streamResponse;

      // Echo every producing (broadcaster) entry.
      // For RTMP broadcasts, producing.token.value is the stream's authKey.
      for (const producing of stream.producing || []) {
        streamResponse.producing.push({
          streamName: producing.streamName,
          token: {
            value: producing.token.value,
            event: producing.token.event,
            stop: streamState.stopped,
            stopReason: streamStopReason,
            // The user id and name would come from your authorization system
            appData: {
              'user.scope': 'broadcaster',
              'user.id': 'broadcaster123',
              'user.name': 'John Doe'
            }
          }
        });
      }

      // Echo every consuming (viewer) entry
      for (const consuming of stream.consuming || []) {
        // Use the token value as the viewer ID
        const viewerId = consuming.token.value;

        // If the viewer state doesn't exist, initialize it
        if (!viewerStates.has(viewerId)) {
          viewerStates.set(viewerId, {
            // Record the viewer's start time
            startTime: now,
            // The viewer is initially not stopped
            stopped: false
          });

          // Schedule the viewer to be stopped after MAX_VIEWER_DURATION
          setTimeout(() => stopViewer(viewerId), MAX_VIEWER_DURATION);
        }

        // Get the current state of the viewer
        const viewerState = viewerStates.get(viewerId);

        streamResponse.consuming.push({
          streamName: consuming.streamName,
          token: {
            value: viewerId,
            event: consuming.token.event,
            stop: viewerState.stopped,
            stopReason: viewerState.stopped ? 'Viewer time limit reached' : undefined,
            appData: {
              'user.scope': 'viewer',
              'user.id': 'viewer456',
              'user.name': 'Jane Smith'
            }
          }
        });
      }
    }
  }

  // Send the response back to Native Frame
  res.json(response);
});

// Start the server on the specified port
const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
  console.log(`Server running on port ${PORT}`);
});

Tracking Viewer Presence​

The same webhook is the platform's only push-based viewer presence signal: there is no separate presence API, but every viewer join and leave arrives as an event on a consuming token. You can extend the handler above into a live viewer roster / concurrent-viewer count:

// streamId -> Set of viewer token values currently watching
const presence = new Map();

function viewersFor(streamId) {
  if (!presence.has(streamId)) presence.set(streamId, new Set());
  return presence.get(streamId);
}

// Inside the consuming loop of the webhook handler:
switch (consuming.token.event) {
  case "joining":       // viewer starts watching
    viewersFor(streamId).add(viewerId);
    break;
  case "closed":        // viewer stopped watching
  case "disconnected":  // viewer connection dropped
    viewersFor(streamId).delete(viewerId);
    break;
  case "polling":       // periodic re-check — doubles as a heartbeat
    viewersFor(streamId).add(viewerId);
    break;
}

console.log(`stream ${streamId}: ${viewersFor(streamId).size} viewers`);

A few practical notes:

  • Presence derived this way is eventually consistent — polling events act as a heartbeat, so a viewer whose closed/disconnected event was missed ages out if you also track a last-seen timestamp per viewer.
  • If you only need a count, you don't need the webhook at all: the stream's manifest document carries a viewCount field, the player's manifest event exposes it client-side, and the platform offers dedicated viewer-count and stream-state endpoints. See Player Events & Liveness.
  • Remember the full-mirror rule: tracking presence doesn't change your response obligations — every consuming entry still has to be echoed back.

Understanding Responsibilities​

When implementing webhook-based stream state management with Native Frame, it's important to understand the division of responsibilities between your custom code and the Native Frame platform. Your primary responsibility lies in implementing and managing the webhook endpoint. This involves setting up a server to receive webhook events, implementing authentication logic to validate tokens and make authorization decisions, and managing the state of streams and viewers based on your specific business logic.

On the other hand, Native Frame takes care of the core platform operations. The platform is responsible for triggering webhook events based on various stream activities, such as when a stream starts or a viewer joins. Upon receiving your webhook responses, Native Frame acts accordingly, managing streams and viewers as instructed. Meanwhile, Native Frame handles all the underlying streaming infrastructure, ensuring smooth and reliable video delivery while you focus on implementing your unique business rules.

By following this guide, you've set up a Node.js server that interacts with Native Frame's webhook system to control stream and viewer states dynamically. This setup allows you to implement custom logic for stream management, providing a tailored experience for your users.