@uploadista/event-emitter-websocket

WebSocket-based event emitter for Uploadista. Sends real-time events to connected clients via persistent WebSocket connections.

Overview

The WebSocket event emitter broadcasts events to connected clients in real-time. Perfect for:

• Real-Time Progress: Stream upload progress to browsers
• Live Notifications: Immediate status updates for users
• Client Subscriptions: Clients subscribe to specific events via WebSocket
• Single-Server Deployments: Works seamlessly with memory broadcaster
• Browser Clients: Native WebSocket support in all browsers

Installation

npm install @uploadista/event-emitter-websocket
# or
pnpm add @uploadista/event-emitter-websocket

Prerequisites

• Node.js 18+ with WebSocket support
• An event broadcaster layer (memory, redis, or ioredis)
• WebSocket client library in browser (built-in)

Quick Start

import { webSocketEventEmitter } from "@uploadista/event-emitter-websocket";
import { memoryEventBroadcaster } from "@uploadista/event-broadcaster-memory";
import { Effect } from "effect";

const program = Effect.gen(function* () {
  // Event emitter is automatically available
});

Effect.runSync(
  program.pipe(
    Effect.provide(webSocketEventEmitter(memoryEventBroadcaster)),
    // ... other layers
  )
);

Features

• ✅ Real-Time Events: Sub-millisecond event delivery to browser
• ✅ Browser Native: Uses standard WebSocket API
• ✅ Connection Management: Automatic cleanup on disconnect
• ✅ Event Routing: Route events to specific subscribers
• ✅ Type Safe: Full TypeScript support
• ✅ Broadcaster Agnostic: Works with any broadcaster

API Reference

Main Exports

webSocketEventEmitter(broadcaster: Layer): Layer<BaseEventEmitterService>

Creates an Effect layer combining WebSocket emitter with a broadcaster.

import { webSocketEventEmitter } from "@uploadista/event-emitter-websocket";
import { memoryEventBroadcaster } from "@uploadista/event-broadcaster-memory";

const layer = webSocketEventEmitter(memoryEventBroadcaster);

WebSocketManager: Service

Manages WebSocket connections and subscriptions.

import { WebSocketManager } from "@uploadista/event-emitter-websocket";
import { Effect } from "effect";

const program = Effect.gen(function* () {
  const manager = yield* WebSocketManager;

  // Add connection
  manager.addConnection("conn-1", connection);

  // Subscribe to events
  manager.subscribeToEvents("upload:123", "conn-1");

  // Emit to subscribers
  manager.emitToEvents("upload:123", "Upload complete");
});

Available Operations

emit(eventKey: string, message: string): Effect<void>

Emit event to all subscribers.

const program = Effect.gen(function* () {
  yield* emitter.emit("upload:123", JSON.stringify({
    status: "completed",
    duration: 5000,
  }));
});

subscribe(eventKey: string, connection: WebSocketConnection): Effect<void>

Subscribe a WebSocket to events.

const program = Effect.gen(function* () {
  yield* emitter.subscribe("upload:123", wsConnection);
  // Client now receives all events for upload:123
});

unsubscribe(eventKey: string): Effect<void>

Unsubscribe from events.

const program = Effect.gen(function* () {
  yield* emitter.unsubscribe("upload:123");
});

Configuration

With Memory Broadcaster

import { webSocketEventEmitter } from "@uploadista/event-emitter-websocket";
import { memoryEventBroadcaster } from "@uploadista/event-broadcaster-memory";
import { uploadServer } from "@uploadista/server";
import { Effect } from "effect";

const program = Effect.gen(function* () {
  const server = yield* uploadServer;
});

Effect.runSync(
  program.pipe(
    Effect.provide(
      webSocketEventEmitter(memoryEventBroadcaster)
    ),
    // ... other layers
  )
);

With Redis Broadcaster

import { webSocketEventEmitter } from "@uploadista/event-emitter-websocket";
import { redisEventBroadcaster } from "@uploadista/event-broadcaster-redis";
import { createClient } from "@redis/client";

const redis = createClient({ url: "redis://localhost:6379" });
const subscriberRedis = createClient({ url: "redis://localhost:6379" });

await redis.connect();
await subscriberRedis.connect();

const layer = webSocketEventEmitter(
  redisEventBroadcaster({ redis, subscriberRedis })
);

Examples

Example 1: Upload Progress to Browser

Server-side:

import { webSocketEventEmitter } from "@uploadista/event-emitter-websocket";
import { memoryEventBroadcaster } from "@uploadista/event-broadcaster-memory";

// Subscribe client to upload events
const subscribeToUpload = (uploadId: string, wsConnection: WebSocket) =>
  Effect.gen(function* () {
    yield* emitter.subscribe(`upload:${uploadId}`, wsConnection);
    console.log(`Client subscribed to ${uploadId}`);
  });

// Emit progress updates
const sendProgress = (uploadId: string, progress: number) =>
  Effect.gen(function* () {
    yield* emitter.emit(`upload:${uploadId}`, JSON.stringify({
      type: "progress",
      progress,
      bytesReceived: Math.floor(100 * 1024 * 1024 * progress),
    }));
  });

Client-side (browser):

const ws = new WebSocket("ws://localhost:3000/uploads/abc123");

// Subscribe to upload
ws.send(JSON.stringify({
  type: "subscribe",
  uploadId: "abc123",
}));

// Receive updates
ws.onmessage = (event) => {
  const message = JSON.parse(event.data);

  if (message.type === "progress") {
    updateProgressBar(message.progress);
    console.log(`Downloaded: ${message.bytesReceived} bytes`);
  }
};

Example 2: Real-Time Flow Status

Server broadcasts flow job status:

const broadcastFlowStatus = (jobId: string, status: string) =>
  Effect.gen(function* () {
    yield* emitter.emit(`flow:${jobId}`, JSON.stringify({
      status,
      timestamp: new Date().toISOString(),
    }));
  });

const trackFlowJob = (jobId: string) =>
  Effect.gen(function* () {
    yield* broadcastFlowStatus(jobId, "queued");
    yield* Effect.sleep("2 seconds");

    yield* broadcastFlowStatus(jobId, "processing");
    yield* Effect.sleep("5 seconds");

    yield* broadcastFlowStatus(jobId, "completed");
  });

Example 3: Multiple Concurrent Uploads

Each client receives only its own upload events:

// Client 1 subscribes to upload A
yield* emitter.subscribe("upload:a", clientA_ws);

// Client 2 subscribes to upload B
yield* emitter.subscribe("upload:b", clientB_ws);

// Events are routed appropriately
yield* emitter.emit("upload:a", JSON.stringify({ progress: 0.5 }));
// Only clientA receives this

yield* emitter.emit("upload:b", JSON.stringify({ progress: 0.3 }));
// Only clientB receives this

Message Format

Standard message structure for events:

interface WebSocketMessage {
  type: string;          // Event type
  payload?: any;         // Event data
  timestamp: string;     // ISO timestamp
  uploadId?: string;     // Associated upload
  flowId?: string;       // Associated flow
}

Example messages:

{
  "type": "progress",
  "payload": { "progress": 0.75, "speed": "2.5 MB/s" },
  "timestamp": "2025-10-21T12:30:45Z",
  "uploadId": "upl_123"
}

Performance

| Operation | Latency | Throughput | |-----------|---------|-----------| | emit() | ~1ms | 1000+ events/sec per connection | | subscribe() | ~100μs | N/A | | Message delivery | ~5-10ms | Depends on network |

Each WebSocket connection can handle 1000+ events per second.

Connection Lifecycle

Client connects
    ↓
subscribe(uploadId, wsConnection)
    ↓
Client receives events via WebSocket
    ↓
unsubscribe(uploadId) or disconnect
    ↓
Cleanup

Best Practices

1. Use Specific Event Keys

// Good: Specific IDs
`upload:${uploadId}`
`flow:${jobId}`
`user:${userId}`

// Avoid: Generic
"updates", "events", "status"

2. Include Type in Message

// Always include type
yield* emitter.emit(`upload:123`, JSON.stringify({
  type: "progress",
  progress: 0.5,
}));

// Client can route based on type
if (msg.type === "progress") {
  updateProgressBar(msg.progress);
}

3. Handle Disconnects

// Server
ws.addEventListener("close", () => {
  // Cleanup subscriptions
  yield* emitter.unsubscribe(`upload:${uploadId}`);
});

// Client
ws.addEventListener("close", () => {
  console.log("Lost connection, retrying...");
  reconnectWebSocket();
});

Deployment

Express + Hono Example

import { createServer } from "http";
import { WebSocketServer } from "ws";
import { webSocketEventEmitter } from "@uploadista/event-emitter-websocket";

const server = createServer();
const wss = new WebSocketServer({ server });

wss.on("connection", (ws) => {
  ws.on("message", (data) => {
    const message = JSON.parse(data.toString());

    if (message.type === "subscribe") {
      const eventKey = `upload:${message.uploadId}`;
      // Subscribe ws to events
    }
  });
});

server.listen(3000);

Troubleshooting

Clients Not Receiving Events

Verify subscription before emitting:

// ✅ Correct order
yield* emitter.subscribe("upload:123", ws);
yield* emitter.emit("upload:123", message);

// ❌ Wrong
yield* emitter.emit("upload:123", message);
yield* emitter.subscribe("upload:123", ws); // Misses event

High Memory Usage

Clean up disconnected connections:

ws.addEventListener("close", () => {
  // Unsubscribe and cleanup
  yield* emitter.unsubscribe(eventKey);
});

Slow Message Delivery

Check broadcaster performance:

// With Redis broadcaster
redis-cli LATENCY LATEST

// With memory broadcaster
// Should be < 1ms

Related Packages

• @uploadista/core - Core types
• @uploadista/event-broadcaster-memory - Memory broadcaster
• @uploadista/event-broadcaster-redis - Redis broadcaster
• @uploadista/server - Upload server
• @uploadista/adapters-hono - Hono WebSocket adapter

License

See LICENSE in the main repository.

See Also

• EVENT_SYSTEM.md - Architecture overview
• WebSocket API - Browser WebSocket
• Server Setup Guide - WebSocket deployment