StreamingKit 🌊

A comprehensive TypeScript streaming library for Node.js supporting multiple stream types for data, media, files, network protocols, and AI/3D applications.

✨ Features

📊 Data Streams

• DataReadStream / DataWriteStream - Generic data streaming
• JsonStream - JSON parsing and serialization
• CsvStream - CSV parsing with headers support
• LineStream - Line-by-line processing
• BatchStream - Batch processing with configurable size

🎬 Media Streams

• MediaStream - Audio/Video streaming
• LiveMediaStream - Real-time streaming with latency control
• OnDemandMediaStream - Seekable media with playback control
• MediaRecorder - Record media streams
• MediaAnalyzer - Analyze FPS, bitrate, and more

📁 File Streams

• FileReadStream / FileWriteStream - File I/O with progress
• ChunkedFileReader - Large file processing
• FileWatchStream - File change monitoring
• DirectoryStream - Directory listing
• FileCopyStream - File copying with progress

🌐 Transport Streams

• HttpStream - HTTP/HTTPS streaming
• SSEStream - Server-Sent Events
• WebSocketStream - WebSocket communication
• WebSocketRPCStream - JSON-RPC over WebSocket
• WebSocketPubSubStream - Pub/Sub pattern
• WebRTCStream - Peer-to-peer streaming
• RTSPStream - RTSP protocol support
• RTMPStream - RTMP protocol support

🤖 AI/3D Streams

• PointCloudStream - LiDAR/3D point cloud data
• AIDetectionStream - Object detection with tracking
• SegmentationStream - Semantic/Instance segmentation
• TelemetryStream - Robot/Vehicle telemetry
• VideoAIStream - Combined video + AI detection

📦 Installation

npm install/streaming-kit

🚀 Quick Start

Data Streaming

import { DataReadStream, JsonStream, CsvStream } from /streaming-kit';

// JSON streaming
const jsonStream = new JsonStream();
jsonStream.on('data', (chunk) => {
  console.log('Parsed JSON chunk:', chunk);
});

// CSV streaming
const csvStream = new CsvStream({ hasHeaders: true });
csvStream.on('record', (record) => {
  console.log('CSV Record:', record);
});
csvStream.write('name,age,city\n');
csvStream.write('John,30,NYC\n');
csvStream.end();

Media Streaming

import { LiveMediaStream, MediaRecorder } from /streaming-kit';

// Create live media stream
const liveStream = new LiveMediaStream({
  video: {
    codec: 'h264',
    width: 1920,
    height: 1080,
    frameRate: 30,
    bitrate: 5000000,
  },
  audio: {
    codec: 'opus',
    sampleRate: 48000,
    channels: 2,
    bitrate: 128000,
  },
  latencyMode: 'realtime',
});

await liveStream.start();

// Record the stream
const recorder = new MediaRecorder(liveStream);
recorder.start();

// Push frames
liveStream.pushFrame({
  type: 'video',
  data: videoFrameData,
  duration: 33,
  isKeyFrame: true,
});

File Streaming

import { FileReadStream, ChunkedFileReader, FileCopyStream } from /streaming-kit';

// Read file with progress
const reader = new FileReadStream({
  path: './large-file.bin',
  highWaterMark: 1024 * 1024, // 1MB chunks
});

reader.on('chunk', (chunk) => {
  console.log(`Progress: ${reader.progress.toFixed(2)}%`);
});

await reader.start();

// Copy file with progress
const copier = new FileCopyStream('./source.zip', './destination.zip');
copier.on('progress', ({ percent }) => {
  console.log(`Copying: ${percent.toFixed(2)}%`);
});
await copier.copy();

WebSocket Streaming

import { WebSocketStream, WebSocketPubSubStream } from /streaming-kit';

// Basic WebSocket
const ws = new WebSocketStream({
  url: 'wss://api.example.com/stream',
  reconnect: true,
  pingInterval: 30000,
});

ws.on('message', (message) => {
  console.log('Received:', message.data);
});

await ws.connect();
await ws.send('Hello, server!');

// Pub/Sub pattern
const pubsub = new WebSocketPubSubStream({
  url: 'wss://pubsub.example.com',
});

await pubsub.connect();

const unsubscribe = pubsub.subscribe('channel:updates', (data) => {
  console.log('Update:', data);
});

await pubsub.publish('channel:updates', { message: 'Hello!' });

WebRTC Streaming

import { WebRTCStream, WebRTCSignaling } from /streaming-kit';

// Peer connection
const peer = new WebRTCStream({
  iceServers: [{ urls: 'stun:stun.l.google.com:19302' }],
});

await peer.initialize();
const dataChannel = peer.createDataChannel('data');

// Create offer
const offer = await peer.createOffer();
// Send offer to remote peer via signaling server

// Receive answer and set it
await peer.setRemoteDescription(answer);

// Send data
peer.send('data', 'Hello via WebRTC!');

RTSP/RTMP Streaming

import { RTSPStream, RTMPStream } from /streaming-kit';

// RTSP client
const rtsp = new RTSPStream({
  url: 'rtsp://camera.local:554/stream',
  transport: 'tcp',
});

rtsp.on('rtp', ({ channel, data }) => {
  console.log(`RTP data on channel ${channel}:`, data.length, 'bytes');
});

await rtsp.start();

// RTMP publishing
const rtmp = new RTMPStream({
  url: 'rtmp://live.twitch.tv/app',
  streamKey: 'your_stream_key',
});

await rtmp.startPublishing();
await rtmp.sendVideo(videoData, timestamp);

AI/3D Streaming

import { PointCloudStream, AIDetectionStream, TelemetryStream } from /streaming-kit';

// Point Cloud (LiDAR)
const pointCloud = new PointCloudStream({
  maxPoints: 100000,
  downsampleRatio: 0.5,
  filterDistance: 50,
});

pointCloud.on('frame', (frame) => {
  console.log(`Point cloud: ${frame.points.length} points`);
});

await pointCloud.start();
pointCloud.pushFrame({
  points: lidarPoints,
  timestamp: Date.now(),
  coordinateSystem: 'cartesian',
});

// AI Object Detection
const detector = new AIDetectionStream({
  confidenceThreshold: 0.7,
  trackingEnabled: true,
  classes: ['person', 'car', 'truck'],
});

detector.on('frame', (frame) => {
  for (const detection of frame.detections) {
    console.log(`${detection.label}: ${(detection.confidence * 100).toFixed(1)}%`);
  }
});

await detector.start();
detector.pushFrame([
  { label: 'person', confidence: 0.95, boundingBox: { x: 100, y: 50, width: 80, height: 200 } },
  { label: 'car', confidence: 0.87, boundingBox: { x: 300, y: 150, width: 200, height: 100 } },
]);

// Telemetry (Robot/Vehicle)
const telemetry = new TelemetryStream({
  sampleRate: 100,
  bufferSize: 10000,
});

telemetry.on('data', (data) => {
  console.log(`Position: (${data.position?.x}, ${data.position?.y}, ${data.position?.z})`);
});

await telemetry.start();
telemetry.pushData({
  position: { x: 10.5, y: 20.3, z: 0.5 },
  orientation: { roll: 0, pitch: 0.1, yaw: 45 },
  velocity: { x: 2.5, y: 0, z: 0 },
  batteryLevel: 85,
  temperature: 42,
});

// Export telemetry to CSV
const csv = telemetry.exportToCSV();

🛠️ Utilities

import {
  pipeline,
  mergeStreams,
  RateLimiter,
  CircularBuffer,
  retry,
  withTimeout,
} from /streaming-kit';

// Pipeline
const processData = pipeline(
  (data: Buffer) => data.toString(),
  (text: string) => JSON.parse(text),
  (obj: any) => obj.value * 2
);

const result = await processData(Buffer.from('{"value": 21}'));
// result = 42

// Rate limiter
const limiter = new RateLimiter(10); // 10 ops/sec
await limiter.acquire();
// Perform rate-limited operation

// Circular buffer
const buffer = new CircularBuffer<number>(100);
buffer.push(1);
buffer.push(2);
const value = buffer.pop(); // 1

// Retry with backoff
const data = await retry(() => fetchData(), {
  maxAttempts: 3,
  delay: 1000,
  backoff: 'exponential',
});

// Timeout wrapper
const result = await withTimeout(longRunningOperation(), 5000, 'Operation timed out');

📚 API Reference

Core Classes

| Class | Description | | --------------------- | ------------------------------ | | BaseStream | Abstract base for all streams | | ReadableStreamBase | Base for readable streams | | WritableStreamBase | Base for writable streams | | DuplexStreamBase | Base for bidirectional streams | | TransformStreamBase | Base for transform streams |

Stream States

• idle - Stream not started
• connecting - Establishing connection
• streaming - Active data flow
• paused - Temporarily stopped
• closed - Stream ended
• error - Error occurred

🧪 Testing

npm test
npm run test:coverage

📄 License

MIT © Jervis Labs

🤝 Contributing

Contributions are welcome! Please read our Contributing Guide for details.

📞 Support

• GitHub Issues
• Documentation