mxrelay-consumer/examples/queue-management.ts

#!/usr/bin/env ts-node

/**
 * Queue management and priority example for SMTP over WebSocket client
 */

import { SMTPOverWSClient, MessagePriority } from '../src/index';

async function queueManagementExample() {
    console.log('Starting SMTP WebSocket Client Queue Management Example\n');

    // Create client - queue management is automatic
    const client = new SMTPOverWSClient({
        url: 'ws://localhost:3000/smtp',
        apiKey: 'your-api-key-here',
        debug: true,
        maxQueueSize: 100,
        reconnectInterval: 2000,
        maxReconnectAttempts: 5
    });

    // Set up comprehensive event monitoring
    client.on('connecting', () => {
        console.log('Connecting to server...');
    });

    client.on('connected', () => {
        console.log('Connected to WebSocket server');
    });

    client.on('authenticated', () => {
        console.log('Authentication successful');
    });

    client.on('queueProcessingStarted', (queueSize) => {
        console.log(`Queue processing started with ${queueSize} messages`);
    });

    client.on('queueProcessingCompleted', (processed, failed) => {
        console.log(`Queue processing completed: ${processed} processed, ${failed} failed`);
    });

    client.on('messageQueued', (messageId, queueSize) => {
        console.log(`Message ${messageId} queued (Queue: ${queueSize})`);
    });

    client.on('messageProcessed', (messageId, responseTime) => {
        console.log(`Message ${messageId} processed in ${responseTime}ms`);
    });

    client.on('messageFailed', (messageId, error) => {
        console.log(`Message ${messageId} failed: ${error.message}`);
    });

    client.on('stateChanged', (oldState, newState) => {
        console.log(`State changed: ${oldState} → ${newState}`);
    });

    try {
        console.log('Demonstrating priority-based queue management...\n');

        // Queue multiple messages with different priorities
        const messagePromises: Promise<string>[] = [];

        // Low priority messages (will be processed last)
        console.log('Queuing low priority messages...');
        for (let i = 1; i <= 3; i++) {
            const promise = client.sendSMTPCommand(`LOW PRIORITY ${i}\r\n`, {
                priority: MessagePriority.LOW,
                timeout: 30000
            });
            messagePromises.push(promise);
        }

        // Normal priority messages
        console.log('Queuing normal priority messages...');
        for (let i = 1; i <= 5; i++) {
            const promise = client.sendSMTPCommand(`NORMAL PRIORITY ${i}\r\n`, {
                priority: MessagePriority.NORMAL,
                timeout: 30000
            });
            messagePromises.push(promise);
        }

        // High priority messages (will be processed first)
        console.log('Queuing high priority messages...');
        for (let i = 1; i <= 2; i++) {
            const promise = client.sendSMTPCommand(`HIGH PRIORITY ${i}\r\n`, {
                priority: MessagePriority.HIGH,
                timeout: 30000
            });
            messagePromises.push(promise);
        }

        // Critical priority message (highest priority)
        console.log('Queuing critical priority message...');
        const criticalPromise = client.sendSMTPCommand('CRITICAL PRIORITY MESSAGE\r\n', {
            priority: MessagePriority.CRITICAL,
            timeout: 30000
        });
        messagePromises.push(criticalPromise);

        console.log(`\nTotal messages queued: ${messagePromises.length}`);
        console.log(`Current queue size: ${client.getQueueSize()}`);

        // Add some messages after a delay to show dynamic queuing
        setTimeout(() => {
            console.log('\nAdding more messages to active queue...');
            
            const additionalPromises = [
                client.sendSMTPCommand('LATE HIGH PRIORITY\r\n', {
                    priority: MessagePriority.HIGH
                }),
                client.sendSMTPCommand('LATE NORMAL PRIORITY\r\n', {
                    priority: MessagePriority.NORMAL
                })
            ];

            messagePromises.push(...additionalPromises);
        }, 2000);

        // Wait for all messages to be processed
        console.log('\nWaiting for all messages to be processed...\n');
        const responses = await Promise.allSettled(messagePromises);

        // Analyze results
        console.log('\nProcessing Results:');
        let successful = 0;
        let failed = 0;

        responses.forEach((result, index) => {
            if (result.status === 'fulfilled') {
                successful++;
                console.log(`Message ${index + 1}: ${result.value.trim()}`);
            } else {
                failed++;
                console.log(`Message ${index + 1}: ${result.reason.message}`);
            }
        });

        console.log(`\nSummary: ${successful} successful, ${failed} failed`);

        // Display final statistics
        const stats = client.getStats();
        console.log('\nFinal Client Statistics:');
        console.log(`   Messages Queued: ${stats.messagesQueued}`);
        console.log(`   Messages Processed: ${stats.messagesProcessed}`);
        console.log(`   Messages Failed: ${stats.messagesFailed}`);
        console.log(`   Average Response Time: ${stats.averageResponseTime.toFixed(2)}ms`);
        console.log(`   Total Connections: ${stats.totalConnections}`);
        console.log(`   Reconnection Attempts: ${stats.reconnectionAttempts}`);
        console.log(`   Connection Uptime: ${(stats.connectionUptime / 1000).toFixed(2)}s`);

    } catch (error) {
        console.error('Error during queue management demo:', error);
    } finally {
        console.log('\nShutting down client...');
        await client.shutdown(10000); // 10 second timeout
        console.log('Client shutdown complete');
    }
}

// Simulate network issues for demonstration
async function resilenceExample() {
    console.log('\nStarting resilience demonstration...\n');

    const client = new SMTPOverWSClient({
        url: 'ws://localhost:3000/smtp',
        apiKey: 'your-api-key-here',
        debug: true,
        reconnectInterval: 1000,
        maxReconnectAttempts: 3,
        messageTimeout: 5000
    });

    client.on('reconnecting', (attempt, maxAttempts) => {
        console.log(`Reconnection attempt ${attempt}/${maxAttempts}`);
    });

    client.on('reconnected', () => {
        console.log('Successfully reconnected');
    });

    client.on('error', (error) => {
        console.log(`Error handled: ${error.message}`);
    });

    try {
        // Queue messages that will trigger reconnection scenarios
        const promises = [];
        
        for (let i = 1; i <= 5; i++) {
            promises.push(
                client.sendSMTPCommand(`RESILIENCE TEST ${i}\r\n`, {
                    retries: 2,
                    timeout: 10000
                }).catch(error => {
                    console.log(`Message ${i} failed: ${error.message}`);
                    return `FAILED: ${error.message}`;
                })
            );
        }

        await Promise.all(promises);
        
    } finally {
        await client.shutdown();
    }
}

// Run the examples
if (require.main === module) {
    (async () => {
        try {
            await queueManagementExample();
            await resilenceExample();
            console.log('\nQueue management examples completed successfully');
        } catch (error) {
            console.error('\nExamples failed:', error);
        } finally {
            process.exit(0);
        }
    })();
}

export { queueManagementExample, resilenceExample };