Event Stream Response

Introduction

A few of our API endpoints use server-sent events (SSE) to provide real-time feedback for long-running operations. This document explains how to consume these event streams in your frontend code.

Basic Implementation Pattern

1. Making the API Request

const response = await axios.post(
  "/api/your-endpoint",
  requestPayload,
  {
    headers: {
      "Content-Type": "application/json",
      "Accept": "text/event-stream",
      "Authorization": `Bearer ${localStorage.getItem("access_token")}`
    },
    responseType: "stream",
    adapter: "fetch"
  }
);

// Get the stream from response
const stream = response.data;
const reader = stream.pipeThrough(new TextDecoderStream()).getReader();

2. Reading and Processing the Stream

let buffer = "";

while (true) {
  const { value, done } = await reader.read();
  
  // Check if stream is complete
  if (done) {
    // Process any remaining data in buffer
    if (buffer.trim()) {
      processBuffer(buffer);
    }
    break;
  }
  
  // Add new data to buffer and process
  buffer += value;
  buffer = processBuffer(buffer);
}

3. Buffer Processing Function

Here's a standard pattern for the buffer processing function:

function processBuffer({
  buffer,
  onEventProcessed,
  onError
}) {
  // Split by double newlines which separate SSE events
  const events = buffer.split("\n\n");
  
  // Keep the last part (potentially incomplete event) in the buffer
  const remainingBuffer = events.pop() || "";
  
  // Process complete events
  for (const event of events) {
    if (!event.trim()) continue;
    
    if (event.includes("event: error")) {
      // Handle error events
      const errorMatch = event.match(/data: (.+)$/m);
      const errorMessage = errorMatch ? errorMatch[1].trim() : "Unknown error";
      onError(errorMessage);
      continue;
    }
    
    // Extract event type and data
    const eventTypeMatch = event.match(/event: (.+)$/m);
    const dataMatch = event.match(/data: (.+)$/m);
    
    if (eventTypeMatch && dataMatch) {
      try {
        const eventType = eventTypeMatch[1].trim();
        const data = JSON.parse(dataMatch[1].trim());
        
        // Process the event via callback
        onEventProcessed({
          event: eventType,
          data: data
        });
      } catch (error) {
        console.error("Error parsing event:", error);
      }
    }
  }
  
  return remainingBuffer;
}

4. Event Handling Using State Updates

// Example of handling events with React state
const updateCompletionStatus = (currentEventName) => {
  // Update UI state based on current event
  setCurrentEvent({
    event: currentEventName,
    data: /* relevant data */
  });
  
  // Update completed steps
  setCompletedSteps((prev) => {
    const uniqueSteps = new Set(prev);
    uniqueSteps.add(currentEventName);
    return Array.from(uniqueSteps);
  });
};

// Use in the buffer processing
buffer = processBuffer({
  buffer,
  onEventProcessed: (data) => {
    updateCompletionStatus(data.event);
  },
  onError: (error) => {
    setError({ message: error });
  }
});

Key Concepts

1. Event Stream Format: Our APIs send events in the Server-Sent Events format with event: and data: fields separated by newlines.

2. Buffer Management: We accumulate incoming data in a buffer and process complete events, keeping any partial events for the next processing cycle.

3. Event Types: Different event types indicate different stages of processing. Handle each according to your UI needs.

4. Error Handling: Error events should be handled appropriately to update UI and inform users.

Common Pitfalls

• Buffer Processing: Always maintain partial events in the buffer to avoid data loss.

• JSON Parsing: Always use try/catch when parsing JSON data from events.

• Memory Management: For very long streams, consider limiting buffer size or clearing processed events.

• Connection Issues: Handle network interruptions and provide retry capabilities.

Example Implementation

Here's a simplified implementation example:

async function processApiWithEvents(endpoint, payload) {
  try {
    setIsLoading(true);
    
    const response = await axios.post(endpoint, payload, {
      headers: {
        "Accept": "text/event-stream",
        "Authorization": `Bearer ${localStorage.getItem("access_token")}`
      },
      responseType: "stream",
      adapter: "fetch"
    });
    
    const stream = response.data;
    const reader = stream.pipeThrough(new TextDecoderStream()).getReader();
    
    let buffer = "";
    
    while (true) {
      const { value, done } = await reader.read();
      if (done) break;
      
      buffer += value;
      buffer = processBuffer({
        buffer,
        onEventProcessed: (eventData) => {
          // Update UI based on event type
          updateProgress(eventData);
        },
        onError: (error) => {
          setError(error);
        }
      });
    }
    
    setIsLoading(false);
    
  } catch (error) {
    setError(error.message);
    setIsLoading(false);
  }
}

This pattern will work for all event stream responses from our APIs.