af-streams

import { DatabaseStream, ConsoleSender } from 'af-streams'; import { type ConnectionOptions as MSSQLConnectionOptions } from 'mssql'; const mssqlConfig: MSSQLConnectionOptions = { user: process.env.DB_USER ?? 'sa', password: process.env.DB_PASSWORD ?? 'YourStrong@Password', server: process.env.DB_HOST ?? 'localhost', database: process.env.DB_NAME ?? 'MyDatabase', options: { encrypt: process.env.DB_ENCRYPT === 'true', // Use for Azure SQL Database, default: false trustServerCertificate: process.env.DB_TRUST_SERVER_CERTIFICATE === 'true' // Change to true for local dev / self-signed certs } }; async function startDatabaseStream() { try { const stream = new DatabaseStream({ dbType: 'mssql', // or 'postgres' dbConfig: mssqlConfig, table: 'YourTableName', timestampField: 'UpdatedAt', primaryKeyFields: ['Id', 'TenantId'], // Composite primary key pollIntervalMs: 5000, // Poll every 5 seconds // Additional options like initialTimestamp, batchSize, etc. }); // Example: Using the ConsoleSender to output streamed data stream.addSender(new ConsoleSender()); // Example: Using a callback sender for custom processing stream.addSender({ send: (data) => { console.log('Callback Sender received record:', data); // Here you could process the data, send to a message queue, etc. } }); console.log('Starting database stream...'); await stream.start(); // To stop the stream after some time or on a specific event // setTimeout(() => { // stream.stop(); // console.log('Database stream stopped.'); // }, 60000); // Stop after 1 minute } catch (error) { console.error('Failed to start database stream:', error); process.exit(1); } } startDatabaseStream();