Huge News!Announcing our $40M Series B led by Abstract Ventures.Learn More
Socket
Sign inDemoInstall
Socket

mux-web-streams

Package Overview
Dependencies
Maintainers
1
Versions
4
Alerts
File Explorer

Advanced tools

Socket logo

Install Socket

Detect and block malicious and high-risk dependencies

Install

mux-web-streams

Multiplex and demultiplex web streams.

  • 1.2.0
  • latest
  • Source
  • npm
  • Socket score

Version published
Weekly downloads
42
increased by223.08%
Maintainers
1
Weekly downloads
 
Created
Source

Mux (and Demux) Web Streams

mux-web-streams enables you to multiplex and demultiplex (AKA "mux" and "demux") streams. Stream multiplexing combines multiple streams into a single stream, so that they can be sent over one communication channel, such as in a single HTTP response. Stream demultiplexing is the opposite operation – it takes a single stream and splits it into multiple streams.

mux-web-streams uses WHATWG-standard Web Streams, which work across Browsers, Node, Bun, Deno.

At Transcend, we use mux-web-streams to stream LLM responses when using langchain on Lamdba functions with Vercel. This allows us to stream responses as they're generated, while also passing other metadata to the client, such as ChainValues.

Installation

You can install mux-web-streams using npm:

npm install mux-web-streams # or `pnpm` or `yarn`

Usage

To use mux-web-streams, import the desired functions from the library:

import { demuxer, muxer } from 'mux-web-streams';

Multiplexing streams

The muxer function is used to multiplex an array of ReadableStreams into a single stream.

import { muxer } from 'mux-web-streams';

// Multiplex readable streams into a single stream
const multiplexedReadableStream: ReadableStream<Uint8Array> = muxer([
  readableStream0,
  readableStream1,
  readableStream2,
  readableStream3,
  readableStream4,
]);

Demultiplexing streams

The demuxer function is used to demultiplex a multiplexed stream back into the original array of ReadableStreams.

import { demuxer } from 'mux-web-streams';

// Demultiplex the stream and listen for emitted streams
const [
  readableStream0,
  readableStream1,
  readableStream2,
  readableStream3,
  readableStream4,
] = demuxer(multiplexedReadableStream, 5);

// Use your streams!
readableStream0.pipeTo(/* ... */);

API

muxer(streams: ReadableStream<SerializableData>[]): ReadableStream<Uint8Array>

Multiplexes an array of ReadableStreams into a single stream.

  • streams: An array of ReadableStreams to be multiplexed.

demuxer(stream: ReadableStream, numberOfStreams: number): ReadableStream<SerializableData>[]

Demultiplexes a single multiplexed ReadableStream into an array of ReadableStreams.

  • stream: The multiplexed stream from muxer().
  • numberOfStreams: The number of streams passed into muxer().

Type SerializableData

The ReadableStreams passed into muxer() must emit SerializableData. This can be a Uint8Array, or anything that's deserializable from JSON, such as string, number, boolean, null, or objects and arrays composed of those primitive types.

Examples

Streaming from a Vercel function

Here's an example using Langchain on Vercel functions. We want to render the AI chat completion as it comes, while also passing the chain values to the client, allowing the end-user to review the source documents behind the chatbot's answer. The chain values return the source documents that were provided to the LLM chat model using a RAG architecture. A more elaborate example might include error messages and other data.

Server-side muxer
// app/api/chat/route.ts
import { muxer } from 'mux-web-streams';

export async function POST(request: Request): Promise<Response> {
  // Get several ReadableStreams
  const chatResponseStream: ReadableStream<string> = createReadableStream();
  const chainValuesStream: ReadableStream<ChainValues> = createReadableStream();

  // Multiplex the streams into a single stream
  const multiplexedStream = muxer([chatResponseStream, chainValuesStream]);

  // Stream the multiplexed data
  return new Response(multiplexedStream, {
    headers: {
      'Content-Type': 'text/event-stream',
      Connection: 'keep-alive',
      'Cache-Control': 'no-cache, no-transform',
    },
  });
}
Client-side demuxer
// components/chat.tsx

import type { ChainValues } from 'langchain/schema';
import { demuxer } from 'mux-web-streams';

export const Chat = () => {
  const [chatResponse, setChatResponse] = useState<string>('');
  const [chainValues, setChainValues] = useState<Record<string, any>>({});

  const onClick = async () => {
    const res = await fetch('/api/chat', {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
      },
      body: JSON.stringify({ myInput: 0 }),
    });

    if (!res.body) {
      throw new Error('No body');
    }

    // Demultiplex the multiplexed stream
    const [chatResponseStream, chainValuesStream] = demuxer(res.body, 2);

    // Write generation to the UI as it comes
    chatResponseStream.pipeThrough(new TextDecoderStream()).pipeTo(
      new WritableStream({
        write(chunk) {
          // Update text of the most recently added element (the AI message)
          setChatResponse(chatResponse + chunk);
        },
      }),
    );

    // Render the chain values when they come
    chainValuesStream.pipeTo(
      new WritableStream({
        write(chunk) {
          setChainValues(JSON.parse(chunk));
        },
      }),
    );
  };

  return (
    <div>
      <button onClick={onClick}>Get multiplexed stream</button>
      <p>Demuxed results:</p>
      <ul>
        <li>Chat response: {chatResponse}</li>
        <li>
          Chain values:{' '}
          <pre>
            <code>{JSON.stringify(chainValues, null, 2)}</code>
          </pre>
        </li>
      </ul>
    </div>
  );
};

License

MIT

Keywords

FAQs

Package last updated on 29 Oct 2023

Did you know?

Socket

Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.

Install

Related posts

SocketSocket SOC 2 Logo

Product

  • Package Alerts
  • Integrations
  • Docs
  • Pricing
  • FAQ
  • Roadmap
  • Changelog

Packages

npm

Stay in touch

Get open source security insights delivered straight into your inbox.


  • Terms
  • Privacy
  • Security

Made with ⚡️ by Socket Inc