rsocket-tcp-server

import { RSocketServer, Payload } from 'rsocket-core'; import { TcpServerTransport } from 'rsocket-tcp-server'; import { Flowable, Single } from 'rsocket-flowable'; const port = 9000; // Define a request handler for the RSocket server const getRequestHandler = ( socket: any, // RSocket instance representing the client connection setupPayload: Payload // Initial payload sent during connection setup ) => { console.log('Client connected:', setupPayload.data?.toString()); return { requestResponse: (payload: Payload): Single<Payload> => { console.log('Received Request-Response:', payload.data?.toString()); return new Single(subscriber => { if (payload.data) { const responseData = `Echo: ${payload.data.toString()}`; subscriber.onComplete({ data: Buffer.from(responseData), metadata: payload.metadata, }); } else { subscriber.onError(new Error('No data in payload')); } }); }, requestStream: (payload: Payload): Flowable<Payload> => { console.log('Received Request-Stream:', payload.data?.toString()); return new Flowable(subscriber => { let count = 0; const interval = setInterval(() => { if (count < 5) { subscriber.onNext({ data: Buffer.from(`Stream ${count++}: ${payload.data?.toString()}`), metadata: payload.metadata, }); } else { clearInterval(interval); subscriber.onComplete(); } }, 1000); return { cancel: () => { console.log('Stream cancelled by client'); clearInterval(interval); }, onComplete: () => { console.log('Stream completed by client'); clearInterval(interval); } }; }); }, // Add other interaction models like fireAndForget, requestChannel as needed }; }; async function startServer() { const server = new RSocketServer({ transport: new TcpServerTransport({ address: '127.0.0.1', port }), getRequestHandler: getRequestHandler, // Setup common serializers for JSON if needed // serializers: JsonSerializers, // For 0.x versions, ensure mime types are compatible setup: { dataMimeType: 'text/plain', metadataMimeType: 'text/plain', keepAlive: 60000, // 60 seconds lifetime: 180000, // 3 minutes }, }); try { await server.bind(); console.log(`RSocket TCP server listening on port ${port}`); } catch (error) { console.error('Failed to start RSocket TCP server:', error); process.exit(1); } } startServer(); // Example: Graceful shutdown process.on('SIGINT', async () => { console.log('Shutting down server...'); // Await server.close() if available in your RSocketServer version process.exit(0); });