mirror of
https://github.com/xtr-dev/rondevu-client.git
synced 2025-12-10 02:43:25 +00:00
feat: v0.9.0 - durable WebRTC connections with automatic reconnection
Major refactor replacing low-level APIs with high-level durable connections.
New Features:
- Automatic reconnection with exponential backoff (1s → 2s → 4s → ... max 30s)
- Message queuing during disconnections
- Durable channels that survive connection drops
- TTL auto-refresh for services (refreshes at 80% of TTL by default)
- Full configuration of timeouts, retry limits, and queue sizes
New API:
- client.exposeService() - Create durable service with automatic TTL refresh
- client.connect() - Create durable connection with automatic reconnection
- client.connectByUuid() - Connect by service UUID
- DurableChannel - Event-based channel wrapper with message queuing
- DurableConnection - Connection manager with reconnection logic
- DurableService - Service manager with TTL auto-refresh
Files Added:
- src/durable/types.ts - Type definitions and enums
- src/durable/reconnection.ts - Exponential backoff utilities
- src/durable/channel.ts - DurableChannel class (358 lines)
- src/durable/connection.ts - DurableConnection class (441 lines)
- src/durable/service.ts - DurableService class (329 lines)
- MIGRATION.md - Comprehensive migration guide
Files Removed:
- src/services.ts - Replaced by DurableService
- src/discovery.ts - Replaced by DurableConnection
BREAKING CHANGES:
- Removed: client.services.*, client.discovery.*, client.createPeer()
- Added: client.exposeService(), client.connect(), client.connectByUuid()
- Handler signature: (channel, peer, connectionId?) → (channel, connectionId)
- Event handlers: .onmessage → .on('message')
- Services: Must call service.start() to begin accepting connections
- Connections: Must call connection.connect() to establish connection
This commit is contained in:
361
src/durable/channel.ts
Normal file
361
src/durable/channel.ts
Normal file
@@ -0,0 +1,361 @@
|
||||
/**
|
||||
* DurableChannel - Message queueing wrapper for RTCDataChannel
|
||||
*
|
||||
* Provides automatic message queuing during disconnections and transparent
|
||||
* flushing when the connection is re-established.
|
||||
*/
|
||||
|
||||
import { EventEmitter } from '../event-emitter.js';
|
||||
import {
|
||||
DurableChannelState
|
||||
} from './types.js';
|
||||
import type {
|
||||
DurableChannelConfig,
|
||||
DurableChannelEvents,
|
||||
QueuedMessage
|
||||
} from './types.js';
|
||||
|
||||
/**
|
||||
* Default configuration for durable channels
|
||||
*/
|
||||
const DEFAULT_CONFIG = {
|
||||
maxQueueSize: 1000,
|
||||
maxMessageAge: 60000, // 1 minute
|
||||
ordered: true,
|
||||
maxRetransmits: undefined
|
||||
} as const;
|
||||
|
||||
/**
|
||||
* Durable channel that survives WebRTC peer connection drops
|
||||
*
|
||||
* The DurableChannel wraps an RTCDataChannel and provides:
|
||||
* - Automatic message queuing during disconnections
|
||||
* - Queue flushing on reconnection
|
||||
* - Configurable queue size and message age limits
|
||||
* - RTCDataChannel-compatible API
|
||||
*
|
||||
* @example
|
||||
* ```typescript
|
||||
* const channel = new DurableChannel('chat', connection, {
|
||||
* maxQueueSize: 500,
|
||||
* maxMessageAge: 30000
|
||||
* });
|
||||
*
|
||||
* channel.on('message', (data) => {
|
||||
* console.log('Received:', data);
|
||||
* });
|
||||
*
|
||||
* channel.on('open', () => {
|
||||
* channel.send('Hello!');
|
||||
* });
|
||||
*
|
||||
* // Messages sent during disconnection are automatically queued
|
||||
* channel.send('This will be queued if disconnected');
|
||||
* ```
|
||||
*/
|
||||
export class DurableChannel extends EventEmitter<DurableChannelEvents> {
|
||||
readonly label: string;
|
||||
readonly config: DurableChannelConfig;
|
||||
|
||||
private _state: DurableChannelState;
|
||||
private underlyingChannel?: RTCDataChannel;
|
||||
private messageQueue: QueuedMessage[] = [];
|
||||
private queueProcessing: boolean = false;
|
||||
private _bufferedAmountLowThreshold: number = 0;
|
||||
|
||||
// Event handlers that need cleanup
|
||||
private openHandler?: () => void;
|
||||
private messageHandler?: (event: MessageEvent) => void;
|
||||
private errorHandler?: (event: Event) => void;
|
||||
private closeHandler?: () => void;
|
||||
private bufferedAmountLowHandler?: () => void;
|
||||
|
||||
constructor(
|
||||
label: string,
|
||||
config?: DurableChannelConfig
|
||||
) {
|
||||
super();
|
||||
this.label = label;
|
||||
this.config = { ...DEFAULT_CONFIG, ...config };
|
||||
this._state = DurableChannelState.CONNECTING;
|
||||
}
|
||||
|
||||
/**
|
||||
* Current channel state
|
||||
*/
|
||||
get readyState(): DurableChannelState {
|
||||
return this._state;
|
||||
}
|
||||
|
||||
/**
|
||||
* Buffered amount from underlying channel (0 if no channel)
|
||||
*/
|
||||
get bufferedAmount(): number {
|
||||
return this.underlyingChannel?.bufferedAmount ?? 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Buffered amount low threshold
|
||||
*/
|
||||
get bufferedAmountLowThreshold(): number {
|
||||
return this._bufferedAmountLowThreshold;
|
||||
}
|
||||
|
||||
set bufferedAmountLowThreshold(value: number) {
|
||||
this._bufferedAmountLowThreshold = value;
|
||||
if (this.underlyingChannel) {
|
||||
this.underlyingChannel.bufferedAmountLowThreshold = value;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Send data through the channel
|
||||
*
|
||||
* If the channel is open, sends immediately. Otherwise, queues the message
|
||||
* for delivery when the channel reconnects.
|
||||
*
|
||||
* @param data - Data to send
|
||||
*/
|
||||
send(data: string | Blob | ArrayBuffer | ArrayBufferView): void {
|
||||
if (this._state === DurableChannelState.OPEN && this.underlyingChannel) {
|
||||
// Channel is open - send immediately
|
||||
try {
|
||||
this.underlyingChannel.send(data as any);
|
||||
} catch (error) {
|
||||
// Send failed - queue the message
|
||||
this.enqueueMessage(data);
|
||||
this.emit('error', error as Error);
|
||||
}
|
||||
} else if (this._state !== DurableChannelState.CLOSED) {
|
||||
// Channel is not open but not closed - queue the message
|
||||
this.enqueueMessage(data);
|
||||
} else {
|
||||
// Channel is closed - throw error
|
||||
throw new Error('Cannot send on closed channel');
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Close the channel
|
||||
*/
|
||||
close(): void {
|
||||
if (this._state === DurableChannelState.CLOSED ||
|
||||
this._state === DurableChannelState.CLOSING) {
|
||||
return;
|
||||
}
|
||||
|
||||
this._state = DurableChannelState.CLOSING;
|
||||
|
||||
if (this.underlyingChannel) {
|
||||
this.underlyingChannel.close();
|
||||
}
|
||||
|
||||
this._state = DurableChannelState.CLOSED;
|
||||
this.emit('close');
|
||||
}
|
||||
|
||||
/**
|
||||
* Attach to an underlying RTCDataChannel
|
||||
*
|
||||
* This is called when a WebRTC connection is established (or re-established).
|
||||
* The channel will flush any queued messages and forward events.
|
||||
*
|
||||
* @param channel - RTCDataChannel to attach to
|
||||
* @internal
|
||||
*/
|
||||
attachToChannel(channel: RTCDataChannel): void {
|
||||
// Detach from any existing channel first
|
||||
this.detachFromChannel();
|
||||
|
||||
this.underlyingChannel = channel;
|
||||
|
||||
// Set buffered amount low threshold
|
||||
channel.bufferedAmountLowThreshold = this._bufferedAmountLowThreshold;
|
||||
|
||||
// Setup event handlers
|
||||
this.openHandler = () => {
|
||||
this._state = DurableChannelState.OPEN;
|
||||
this.emit('open');
|
||||
|
||||
// Flush queued messages
|
||||
this.flushQueue().catch(error => {
|
||||
this.emit('error', error);
|
||||
});
|
||||
};
|
||||
|
||||
this.messageHandler = (event: MessageEvent) => {
|
||||
this.emit('message', event.data);
|
||||
};
|
||||
|
||||
this.errorHandler = (event: Event) => {
|
||||
this.emit('error', new Error(`Channel error: ${event.type}`));
|
||||
};
|
||||
|
||||
this.closeHandler = () => {
|
||||
if (this._state !== DurableChannelState.CLOSING &&
|
||||
this._state !== DurableChannelState.CLOSED) {
|
||||
// Unexpected close - transition to connecting (will reconnect)
|
||||
this._state = DurableChannelState.CONNECTING;
|
||||
}
|
||||
};
|
||||
|
||||
this.bufferedAmountLowHandler = () => {
|
||||
this.emit('bufferedAmountLow');
|
||||
};
|
||||
|
||||
// Attach handlers
|
||||
channel.addEventListener('open', this.openHandler);
|
||||
channel.addEventListener('message', this.messageHandler);
|
||||
channel.addEventListener('error', this.errorHandler);
|
||||
channel.addEventListener('close', this.closeHandler);
|
||||
channel.addEventListener('bufferedamountlow', this.bufferedAmountLowHandler);
|
||||
|
||||
// If channel is already open, trigger open event
|
||||
if (channel.readyState === 'open') {
|
||||
this.openHandler();
|
||||
} else if (channel.readyState === 'connecting') {
|
||||
this._state = DurableChannelState.CONNECTING;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Detach from the underlying RTCDataChannel
|
||||
*
|
||||
* This is called when a WebRTC connection drops. The channel remains alive
|
||||
* and continues queuing messages.
|
||||
*
|
||||
* @internal
|
||||
*/
|
||||
detachFromChannel(): void {
|
||||
if (!this.underlyingChannel) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Remove event listeners
|
||||
if (this.openHandler) {
|
||||
this.underlyingChannel.removeEventListener('open', this.openHandler);
|
||||
}
|
||||
if (this.messageHandler) {
|
||||
this.underlyingChannel.removeEventListener('message', this.messageHandler);
|
||||
}
|
||||
if (this.errorHandler) {
|
||||
this.underlyingChannel.removeEventListener('error', this.errorHandler);
|
||||
}
|
||||
if (this.closeHandler) {
|
||||
this.underlyingChannel.removeEventListener('close', this.closeHandler);
|
||||
}
|
||||
if (this.bufferedAmountLowHandler) {
|
||||
this.underlyingChannel.removeEventListener('bufferedamountlow', this.bufferedAmountLowHandler);
|
||||
}
|
||||
|
||||
this.underlyingChannel = undefined;
|
||||
this._state = DurableChannelState.CONNECTING;
|
||||
}
|
||||
|
||||
/**
|
||||
* Enqueue a message for later delivery
|
||||
*/
|
||||
private enqueueMessage(data: string | Blob | ArrayBuffer | ArrayBufferView): void {
|
||||
// Prune old messages first
|
||||
this.pruneOldMessages();
|
||||
|
||||
const message: QueuedMessage = {
|
||||
data,
|
||||
enqueuedAt: Date.now(),
|
||||
id: `msg-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`
|
||||
};
|
||||
|
||||
this.messageQueue.push(message);
|
||||
|
||||
// Handle overflow
|
||||
const maxQueueSize = this.config.maxQueueSize ?? 1000;
|
||||
if (this.messageQueue.length > maxQueueSize) {
|
||||
const excess = this.messageQueue.length - maxQueueSize;
|
||||
this.messageQueue.splice(0, excess);
|
||||
this.emit('queueOverflow', excess);
|
||||
console.warn(
|
||||
`DurableChannel[${this.label}]: Dropped ${excess} messages due to queue overflow`
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Flush all queued messages through the channel
|
||||
*/
|
||||
private async flushQueue(): Promise<void> {
|
||||
if (this.queueProcessing || !this.underlyingChannel ||
|
||||
this.underlyingChannel.readyState !== 'open') {
|
||||
return;
|
||||
}
|
||||
|
||||
this.queueProcessing = true;
|
||||
|
||||
try {
|
||||
// Prune old messages before flushing
|
||||
this.pruneOldMessages();
|
||||
|
||||
// Send all queued messages
|
||||
while (this.messageQueue.length > 0) {
|
||||
const message = this.messageQueue.shift();
|
||||
if (!message) break;
|
||||
|
||||
try {
|
||||
this.underlyingChannel.send(message.data as any);
|
||||
} catch (error) {
|
||||
// Send failed - re-queue message
|
||||
this.messageQueue.unshift(message);
|
||||
throw error;
|
||||
}
|
||||
|
||||
// If buffer is getting full, wait for it to drain
|
||||
if (this.underlyingChannel.bufferedAmount > 16 * 1024 * 1024) { // 16MB
|
||||
await new Promise<void>((resolve) => {
|
||||
const checkBuffer = () => {
|
||||
if (!this.underlyingChannel ||
|
||||
this.underlyingChannel.bufferedAmount < 8 * 1024 * 1024) {
|
||||
resolve();
|
||||
} else {
|
||||
setTimeout(checkBuffer, 100);
|
||||
}
|
||||
};
|
||||
checkBuffer();
|
||||
});
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
this.queueProcessing = false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove messages older than maxMessageAge from the queue
|
||||
*/
|
||||
private pruneOldMessages(): void {
|
||||
const maxMessageAge = this.config.maxMessageAge ?? 60000;
|
||||
if (maxMessageAge === Infinity || maxMessageAge <= 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
const now = Date.now();
|
||||
const cutoff = now - maxMessageAge;
|
||||
|
||||
const originalLength = this.messageQueue.length;
|
||||
this.messageQueue = this.messageQueue.filter(msg => msg.enqueuedAt >= cutoff);
|
||||
|
||||
const pruned = originalLength - this.messageQueue.length;
|
||||
if (pruned > 0) {
|
||||
console.warn(
|
||||
`DurableChannel[${this.label}]: Pruned ${pruned} old messages (older than ${maxMessageAge}ms)`
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the current queue size
|
||||
*
|
||||
* @internal
|
||||
*/
|
||||
getQueueSize(): number {
|
||||
return this.messageQueue.length;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user