node.re

module EventEmitter = {
  module Event = {
    type t('e, 'v) = ..;
    type t(_, _) +=
      | NewListener : t([>`EventEmitter], string) 
      | RemoveListener : t([>`EventEmitter], string);
    type pkg =
      | Pkg(t('e, 'v)): pkg;
  };
  module Listener = {
    type t('v) = 'v => unit;
    type pkg =
      | Pkg(t('v)): pkg;
    [@bs.get] external name : t('v) => string = "name";
  };
  class type t ('e) =
    [@bs]
    {
      pub addListener: (Event.t('e, 'v), Listener.t('v)) => Js.t(t('e));
      pub emit: (Event.t('e, 'v), 'v) => Js.t(t('e));
      pub eventNames: unit => array(Event.pkg);
      pub getMaxListeners: unit => int;
      pub listenerCount: Event.t('e, 'v) => int;
      pub listeners: Event.t('e, 'v) => array(Listener.t('v));
      pub on: (Event.t('e, 'v), Listener.t('v)) => Js.t(t('e));
      pub once: (Event.t('e, 'v), Listener.t('v)) => Js.t(t('e));
      pub prependListener: (Event.t('e, 'v), Listener.t('v)) => Js.t(t('e));
      pub prependOnceListener: (Event.t('e, 'v), Listener.t('v)) => Js.t(t('e));
      pub removeAllListeners: (Event.t('e, 'v), Listener.t('v)) => Js.t(t('e));
      pub removeListeners: (Event.t('e, 'v), Listener.t('v)) => Js.t(t('e));
      pub setMaxListeners: int => Js.t(t('e))
    };
};

module Stream = {
  type ns = [`Stream | `EventEmitter];
  type EventEmitter.Event.t(_, _) +=
    | Close : EventEmitter.Event.t(ns, unit)
    | Error : EventEmitter.Event.t(ns, unit);
  class type t ('e) =
    [@bs]
    {
      inherit EventEmitter.t ('e);
      pub destroy: unit => unit;
    };
};

module ReadableStream = {
  type ns = [ `ReadableStream | Stream.ns ];
  type EventEmitter.Event.t(_, _) +=
    | Data : EventEmitter.Event.t(ns, Node_buffer.t)
    | End  : EventEmitter.Event.t(ns, unit);
  class type t ('e) =
    [@bs]
    {
      inherit Stream.t (ns);
      pub pipe: Js.t(Stream.t([>`WritableStream])) => unit;
    };
};

module WritableStream = {
  type ns = [ `WritableStream | Stream.ns ];
  type EventEmitter.Event.t(_, _) +=
    | Pipe : EventEmitter.Event.t(ns, Js.t(Stream.t([>`ReadableStream])));
  class type t ('e) =
    [@bs]
    {
      inherit Stream.t (ns);
    };
};

[@bs.module "fs"] external createReadStream : string => Js.t(ReadableStream.t([`WritableStream])) = "";
[@bs.module "fs"] external createWriteStream : string => Js.t(WritableStream.t([`ReadableStream])) = "";

Works if called like this:

open Node;

let rs = createReadStream("./test.txt");
let ws = createWriteStream("./test2.txt");
ws##on(WritableStream.Pipe, (src) => {
    log("got piped!");
    Js.log(src);
});
rs##pipe(ws);