blog/real-world-observables

ftp-1.js

class Ftp {
  constructor(host = "localhost", port = 21) {
    this._ftpSocket = Net.createConnection(port, host);
    this._ftpSocket.setEncoding("utf8");

    this.responses = fromStream(this._ftpSocket);
  }
}

ftp-repl.txt

$ node client.js
ftp#localhost:21> user sergi
331 User sergi accepted, provide password.
ftp#localhost:21> pass p4ssw0rd
230 User sergi logged in.
ftp#localhost:21> cwd ~/projects/simple_ftp
250 CWD command successful.
ftp#localhost:21> stat .
211-status of .:

total 1
-rw-r--r--    1 sergi  staff    44 Dec  3 12:59 .babelrc
drwxr-xr-x   13 sergi  staff   442 Apr 26 12:44 .git
-rw-r--r--    1 sergi  staff  1294 Jan 21 11:26 Makefile
drwxr-xr-x    5 sergi  staff   170 Jan 22 15:34 build
-rw-r--r--    1 sergi  staff    69 Nov 20 16:22 jsconfig.json
drwxr-xr-x    4 sergi  staff   136 Jan 22 16:02 lib
drwxr-xr-x  264 sergi  staff  8976 Apr 26 12:42 node_modules
-rw-r--r--    1 sergi  staff   428 Jan 21 13:13 package.json
drwxr--r--    4 sergi  staff   136 Nov 20 16:24 typings
211 End of Status
ftp#localhost:21> feat
211-Features supported

MDTM
MLST Type*;Size*;Modify*;Perm*;Unique*;
REST STREAM
SIZE
TVFS
211 End
ftp#localhost:21>

ftp-responses.txt

150-This is the first line of a mark
123-This line does not end the mark; note the hyphen
150 This line ends the mark
226-This is the first line of the second response
226 This line does not end the response; note the leading space
226 This is the last line of the response, using code 226

ftp-session.txt

$ telnet ftp.kernel.org 21
Trying 198.145.20.140...
Connected to ftp.all.kernel.org.
Escape character is '^]'.
220 Welcome to kernel.org
user anonymous
331 Please specify the password.
pass anonymous
230 Login successful.
cwd pub/linux
250 Directory successfully changed.
stat .
213-Status follows:
drwxr-xr-x   14 ftp      ftp          4096 Nov 11  2014 .
drwxr-xr-x    9 ftp      ftp          4096 Dec 01  2011 ..
drwxrwxr-x    7 ftp      ftp          4096 Nov 10  2002 daemons
drwxrwxr-x    5 ftp      ftp          4096 Mar 03  2001 devel
drwxrwxr-x    5 ftp      ftp          4096 Nov 19  2007 docs
drwxr-xr-x   24 ftp      ftp          4096 Feb 23  2015 kernel
drwxr-xr-x   13 ftp      ftp          4096 Jan 03  2012 libs
drwxr-xr-x    6 ftp      ftp          4096 Aug 07  2014 network
drwxr-xr-x    3 ftp      ftp          4096 Jan 23  2011 status
drwxr-xr-x   21 ftp      ftp          4096 Jun 17  2015 utils
213 End of status

modules.js

// We use Node sockets to connect to the FTP server
import Net from "net";
import Rx from "rx";
// Convert Observables from and to streams, which are
// the interface that Node.js sockets implement
import { fromStream, writeToStream } from "rx-node";

parseFtpResponses.js

const RE_RES = /^(\d\d\d)\s.*/;
const RE_MULTI = /^(\d\d\d)-/;

function parseFtpResponses(prev, cur) {
  let response = '';
  let accumulated = prev.accumulated;

  if (RE_MULTI.test(accumulated)) {
    if (RE_RES.test(cur)) {
      response = accumulated + '\n' + cur;
      accumulated = '';
    } else {
      accumulated = accumulated + '\n' + cur;
    }
  } else if (RE_MULTI.test(cur)) {
    accumulated = cur;
  } else {
    response = cur;
  }

  return { response, accumulated };
}

repl.js

const repl = require("repl");
const Rx = require("rx");
const Ftp = require("./simple_ftp"); // Our library!
const client = new Ftp();

let responses = client.tuples.map(t => t[1]);
// This Subject will emit
let callbacks = new Rx.Subject();

repl.start({
  prompt: "ftp#" + client.host + ":" + client.port + "> ",
  input: process.stdin,
  output: process.stdout,
  eval: (cmd, context, filename, callback) => {
    client.command(cmd);
    callbacks.onNext(callback);
  }
});

responses.zip(callbacks).subscribe(result => {
  let [cmdResult, callback] = result;
  callback(cmdResult);
});

responses.js

this.responses = fromStream(this._ftpSocket)
  .flatMap(res => {
    // Splits both \n and \r\n cases
    let lines = res.split("\n").map(line => line.trim());
    return Rx.Observable.from(lines);
  })
  .scan(parseFtpResponses, { command: "", accumulated: "" })
  .filter(value => value.command.length > 0)
  .map(value => value.command);