AWS

AWS Lambda

Nodejs

Rust Programming Language

Golang

Server-Sent Events with AWS Lambda response streaming

Johannes Geiger

5 min read

·

Jun 28, 2024

--

3

Listen

Share

Introduction

Even though various AWS services (AppSync, IoT Core, API Gateway) support WebSockets as asynchronous communication mechanism more or less out of the box sometimes it’s sufficient to only have a unidirectional asynchronous channel from the server to the client — and that’s where Server-Sent Events (SSE) come into play. Previously, running SSE in AWS required us to run our own SSE server in e.g. Fargate but recently AWS Lambda response streaming opened up the possibility to use AWS Lambda for SSE as both response streaming allows us to return data to clients over a longer period of time and the JavaScript EventSource automatically reconnects if the connection is being closed by the server — so, having a Lambda that serves SSE for e.g. 10 seconds and then closes the connection followed by the client reconnecting allows for an almost seamless SSE experience without the danger of Lambdas timing out (if the timeout is set high enough of course). The only downside is that response streaming currently won’t work for Lambda behind an Application Load Balancer or API Gateway and only through direct access via Function URL. Further, response streaming is currently only supported in the Node.js runtime (and custom runtimes) but neither Python nor Java for example.

For this case study I am considering Node.js Lambdas (for the built-in support) and Go/Rust (for the custom runtime support). The complete code examples are all added in an appendix.

Update (20 November 2025)

Yesterday, AWS announced that REST API now support response streaming which solves one caveat above: That response streaming was only available via a Lambdas’ Function URL, because with that and us having added a custom domain name to our REST API we can make our Server Sent Event functionality comfortably available to our consumers.

Node.js

The biggest difference between “normal” Lambdas and response streaming Lambdas is the definition of the handler:

export const handler = awslambda.streamifyResponse(async (event, responseStream) => {
  // ...
});

and the easiest way to stream data asynchronously is to use a pipeline:

import util from 'util';
import stream from 'stream';

const pipeline = util.promisify(stream.pipeline);

What makes this now possible to use for SSE is that pipelines support asynchronous generators using yield to return the data followed by a delay to simulate a constant stream of data:

const asyncGenerator = async function* () {
  for (let i = 0; i < 10; i++) {
    yield `data: Hello World!\n\n`;
    await new Promise(r => setTimeout(r, 1000));
  }
};

Finally, if we’re now declaring the response with the correct MIME type:

export const handler = awslambda.streamifyResponse(async (event, responseStream) => {
  // ...

  const httpResponseMetadata = {
    statusCode: 200,
    headers: {
      "Content-Type": "text/event-stream"
      }
    };
  const httpResponseStream = awslambda.HttpResponseStream.from(responseStream, httpResponseMetadata);
}

we can use the Lambda’s Function URL directly in JavaScript:

const eventSource = new EventSource("...");
eventSource.onmessage = (event) => alert(JSON.stringify(event));

Go

Similar to Node.js the handler itself changes, in Go it’s now using a different return type:

func HandleRequest(_ context.Context, _ *events.LambdaFunctionURLRequest) (*events.LambdaFunctionURLStreamingResponse, error) {
  return &events.LambdaFunctionURLStreamingResponse{
    StatusCode: 200,
    Headers: map[string]string{
      `Content-Type`: `text/event-stream`,
    },
    Body: ServerSentEventReader{
    },
  }, nil
}

and similar to Node.js we can now treat the body (an implementation of io.Reader) as if it’s asynchronous, we just have to ensure that after reaching the number of requests our reader returns io.EOF to mark the end of the stream which subsequently closes the connection:

type ServerSentEventReaderState struct {
  currentRequest int
}

var state = ServerSentEventReaderState{}

const Response = "data: Hello World!\n\n"

type ServerSentEventReader struct {
}

func (r ServerSentEventReader) Read(b []byte) (int, error) {
  if state.currentRequest == 10 {
    state.currentRequest = 0
    return 0, io.EOF
  }

  time.Sleep(1 * time.Second)
  n := copy(b, Response)
  state.currentRequest++

  return n, nil
}

Rust

Rust uses a similar mechanism as Node.js and Go by returning a new type of response but makes use of channels that act as the object to be returned that can get data pushed into with delay:

async fn handler(_event: LambdaEvent<Value>) -> Result<Response<Body>, Error> {
  let (mut tx, rx) = channel();

  // ...

  let mut header_map = HeaderMap::new();
  header_map.insert("Content-Type", "text/event-stream".parse().unwrap());

  Ok(StreamResponse {
    metadata_prelude: MetadataPrelude {
      status_code: Default::default(),
      headers: HeaderMap::from(header_map),
      cookies: Default::default(),
    },
    stream: rx,
  })
}

and the source of data can then be e.g. a child thread that makes use of the tx sender:

tokio::spawn(async move {
  for _ in 0..10 {
    match tx.send_data(("data: Hello World\n\n").into()).await {
      Ok(()) => {}
      Err(e) => eprintln!("An error occurred: {}", e),
    }
    task::sleep(Duration::from_millis(1000)).await;
  }
});

Even though those experiments aren’t as seamless as the existing AWS services for WebSockets they demonstrate how simple SSE can be implemented with existing AWS functionality and it remains to be seen if AWS improves that even further to simplify SSE running in AWS.

Appendix

index.mjs:

import util from 'util';
import stream from 'stream';

const count = process.env.COUNT ?? 5;
const delayMs = process.env.DELAY_MS ?? 1000;

const pipeline = util.promisify(stream.pipeline);

export const handler = awslambda.streamifyResponse(async (event, responseStream) => {
    const asyncGenerator = async function* () {
        console.log(`Request received`);

        for (let i = 0; i < count; i++) {
            const response = `data: Hello World!\n\n`;
            console.log(`Iteration ${i}: ${response}`);
            yield response;
            await new Promise(r => setTimeout(r, delayMs));
        }

        console.log(`${count} iterations reached, closing connection.`);
    };

    const httpResponseMetadata = {
        statusCode: 200,
        headers: {
            "Content-Type": "text/event-stream"
        }
    };
    const httpResponseStream = awslambda.HttpResponseStream.from(responseStream, httpResponseMetadata);

    await pipeline(asyncGenerator, httpResponseStream);
});

main.go:

package main

import (
  "context"
  "github.com/aws/aws-lambda-go/events"
   "github.com/aws/aws-lambda-go/lambda"
   "io"
   "log"
   "time"
)

type ServerSentEventReaderState struct {
   currentRequest int
}

var logger = log.Default()
var state = ServerSentEventReaderState{}

const Response = "data: Hello World!\n\n"

type ServerSentEventReader struct {
  requests int
  delay    time.Duration
}

func (r ServerSentEventReader) Read(b []byte) (int, error) {
  if state.currentRequest == r.requests {
    logger.Printf(`%d iterations reached, closing connection.`, r.requests)
    state.currentRequest = 0
    return 0, io.EOF
  }

  time.Sleep(r.delay * time.Second)
  n := copy(b, Response)
  logger.Printf(`Iteration %d: %s`, state.currentRequest, Response)
  state.currentRequest++

  return n, nil
}

func HandleRequest(_ context.Context, _ *events.LambdaFunctionURLRequest) (*events.LambdaFunctionURLStreamingResponse, error) {
  logger.Println(`Request received`)

  return &events.LambdaFunctionURLStreamingResponse{
    StatusCode: 200,
    Headers: map[string]string{
      `Content-Type`: `text/event-stream`,
    },
    Body: ServerSentEventReader{
      requests: 10,
      delay:    time.Duration(2),
    },
  }, nil
}

func main() {
  lambda.Start(HandleRequest)
}

main.rs:

use std::time::Duration;

use async_std::task;
use http::HeaderMap;
use lambda_runtime::{Error, LambdaEvent, MetadataPrelude, service_fn, streaming::{Body, channel}, StreamResponse};
use lambda_runtime::streaming::Response;
use serde_json::Value;

async fn handler(_event: LambdaEvent<Value>) -> Result<Response<Body>, Error> {
    let (mut tx, rx) = channel();

    tokio::spawn(async move {
        for _ in 0..10 {
            match tx.send_data(("data: Hello World\n\n").into()).await {
                Ok(()) => {}
                Err(e) => eprintln!("An error occurred: {}", e),
            }
            task::sleep(Duration::from_millis(1000)).await;
        }
    });

    let mut header_map = HeaderMap::new();
    header_map.insert("Content-Type", "text/event-stream".parse().unwrap());

    Ok(StreamResponse {
        metadata_prelude: MetadataPrelude {
            status_code: Default::default(),
            headers: HeaderMap::from(header_map),
            cookies: Default::default(),
        },
        stream: rx,
    })
}

#[tokio::main]
async fn main() -> Result<(), Error> {
    lambda_runtime::run(service_fn(handler)).await?;
    Ok(())
}