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. Inside this endpoint, extract the programs object from the request body, which contains the data about programs and streams that need to be processed:

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

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

  // We'll fill this in the next steps

  res.json(response);
});

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) => {
  // Extract the programs object from the webhook request
  const { programs } = req.body;

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

  // Initialize the response object
  const response = { programs: {} };

  // Iterate over each program in the request
  for (const [programId, program] of Object.entries(programs)) {
    response.programs[programId] = {
      // By default, do not stop the program
      stop: false,
      // Require authentication for the program
      needAuth: true,
      // Initialize the streams object
      streams: {}
    };

    // 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);

      // Prepare appData with broadcaster information
      // The user id and name would come from your authorization system
      const appData = {
        "user.scope": "broadcaster",
        "user.id": "broadcaster123",
        "user.name": "John Doe"
      };

      // Build the response for this stream
      response.programs[programId].streams[streamId] = {
        // Require authentication for the stream
        needAuth: true,
        // Indicate whether the stream should be stopped
        stop: streamState.stopped,
        // Provide a stop reason if stopped
        stopReason: streamState.stopped ? "Stream duration limit reached" : undefined,
        // Include the stream token if available
        token: stream.token ? stream.token.value : undefined,
        // Initialize the viewTokens object
        viewTokens: {},
        // Attach broadcaster appData
        appData
      };

      // Handle viewer tokens if present
      if (stream.viewTokens) {
        for (const viewToken of stream.viewTokens) {
          // Use the token value as the viewer ID
          const viewerId = viewToken.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,
              appData: {
                "user.scope": "viewer",
                "user.id": "viewer456",
                "user.name": "Jane Smith"
              }
            });

            // 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);

          // Build the response for this viewer
          response.programs[programId].streams[streamId].viewTokens[viewerId] = {
            // Indicate whether the viewer should be stopped
            stop: viewerState.stopped,
            // Provide a stop reason if stopped
            stopReason: viewerState.stopped ? "Viewer time limit reached" : undefined,
            // Attach viewer appData
            appData: viewerState.appData
          };
        }
      }
    }
  }

  // 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, 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 object that mirrors the structure expected by Native Frame. This response includes whether to stop the program or stream, authentication requirements, and any additional data (appData) about the broadcaster and viewers. By providing detailed information in the response, we instruct Native Frame on how to handle each stream and viewer.

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 4: 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`);
  }
}

When Native Frame receives the stopped property in the webhook response, it will stop the 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) => {
  // Extract the programs object from the webhook request
  const { programs } = req.body;

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

  // Initialize the response object
  const response = { programs: {} };

  // Iterate over each program in the request
  for (const [programId, program] of Object.entries(programs)) {
    response.programs[programId] = {
      // By default, do not stop the program
      stop: false,
      // Require authentication for the program
      needAuth: true,
      // Initialize the streams object
      streams: {}
    };

    // 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);

      // Prepare appData with broadcaster information
      // The user id and name would come from your authorization system
      const appData = {
        "user.scope": "broadcaster",
        "user.id": "broadcaster123",
        "user.name": "John Doe"
      };

      // Build the response for this stream
      response.programs[programId].streams[streamId] = {
        // Require authentication for the stream
        needAuth: true,
        // Indicate whether the stream should be stopped
        stop: streamState.stopped,
        // Provide a stop reason if stopped
        stopReason: streamState.stopped ? "Stream duration limit reached" : undefined,
        // Include the stream token if available
        token: stream.token ? stream.token.value : undefined,
        // Initialize the viewTokens object
        viewTokens: {},
        // Attach broadcaster appData
        appData
      };

      // Handle viewer tokens if present
      if (stream.viewTokens) {
        for (const viewToken of stream.viewTokens) {
          // Use the token value as the viewer ID
          const viewerId = viewToken.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,
              appData: {
                "user.scope": "viewer",
                "user.id": "viewer456",
                "user.name": "Jane Smith"
              }
            });

            // 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);

          // Build the response for this viewer
          response.programs[programId].streams[streamId].viewTokens[viewerId] = {
            // Indicate whether the viewer should be stopped
            stop: viewerState.stopped,
            // Provide a stop reason if stopped
            stopReason: viewerState.stopped ? "Viewer time limit reached" : undefined,
            // Attach viewer appData
            appData: viewerState.appData
          };
        }
      }
    }
  }

  // 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}`);
});

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.