Reactive TCP socket programming in Node.js

tcp-socket.ts

import { connect, NetConnectOpts } from 'net';
import { Observable, Observer, Subject } from 'rxjs';

// TCP Client
const fromSocket = (options: NetConnectOpts): Subject<Buffer> => {
  const socket = connect(options);

  const observable = Observable.create(function(obs) {
    socket.on('data', obs.next.bind(obs));
    socket.on('error', obs.error.bind(obs));
    socket.on('close', obs.complete.bind(obs));

    return socket.destroy.bind(socket);
  });

  const observer: Partial<Observer<Buffer>> = {
    next: data => socket.write(data)
  }

  const subject = Subject.create(observer, observable);
  return subject;
}

// Usage example
const host = 'localhost';
const port = 55013;

const socket = fromSocket({ host, port })

const connection = socket.subscribe({
  next: buffer => console.log(buffer.toString()),
  error: (err) => console.log(`error occurred: ${err.message}`),
  complete: () => console.log('done')
});

socket.next(Buffer.from('Hello World!!!'));
  

Thanks! Just what I was looking for 👏

Any idea how you would achieve this with the latest version of RxJs?

Subject.create() in unfortunately deprecated 😔