Package Exports

rxjs-shell
rxjs-shell/dist/index.js

This package does not declare an exports field, so the exports above have been automatically detected and optimized by JSPM instead. If any package subpath is missing, it is recommended to post an issue to the original package (rxjs-shell) to support the "exports" field. If that is not possible, create a JSPM override to customize the exports field for this package.

Readme

rxjs-shell

rxjs operators for execute shell command with ease.

Features

Wrap nodejs asynchronous process creation methods to rxjs Observable.
Kill child process when unsubscribed.
Use subject to communicate with child process.

Functions

exec(command[, options][, proccallback]) → Observable<{stdout: string | Buffer, stderr: string | Buffer}>

options interface is same with nodejs exec method
procCallback you can pass function. ChildProcess will be passed first argument.

import {exec} from 'rxjs-shell';

exec('echo Hello World').subscribe(output => {
  console.log(output.stdout.toString('utf8')); // Hello World\n
});


// using `procCallback`
exec('cat -', undefined, proc => {
  proc.stdin?.write('Hello World');
  proc.stdin?.end(); // it may cause endless process if you don't handle right.
}).subscribe(output => { /* ... */ })

execFile(file[, args][, options]) → Observable<{stdout: string | Buffer, stderr: string | Buffer}>

options interface is same with nodejs execFile method

import {existSync} from 'fs';
import {execFile} from 'rxjs-shell';
execFile('./touchFile.sh').subscribe(() => {
  console.log(existSync('touched.txt')); // true
});

spawn(command[, args][, options][, procCallback]) → Observable<{type: 'stdout' | 'stderr', chunk: Buffer}>

spawn emits stdout, stderr's buffer from command execution.
options interface is same with nodejs spawn method
procCallback you can pass function. ChildProcessWithoutNullStreams will be passed first argument.

import {spawn} from 'rxjs-shell';

spawn('git clone http://github.com/johnny-mh/rxjs-shell-operators')
  .pipe(tap(chunk => process.stdout.write(String(chunk.chunk))))
  .subscribe();

// using `procCallback`
spawn('cat', ['-'], undefined, proc => {
  proc.stdin.write('hello world');
  proc.stdin.end(); // caution
}).subscribe(output => { /* ... */ });

fork(modulePath[, args][, options]) → Observable<Serializable>

same with spawn but have own options interface that extend nodejs's fork options to communicate with child process.

import {Subject} from 'rxjs';
import {fork} from 'rxjs-shell';

const send = new Subject<string>();

fork('echo.js', undefined, {send}).subscribe(msgFromChildProc =>
  console.log(msgFromChildProc)
);

send.next('message to child process');

Operators

trim(encoding = 'utf8')

trim child process output

import {exec, trim} from 'rxjs-shell';

exec('echo Hello').subscribe(output => console.log(output.stdout.toString())); // Hello\n

exec('echo Hello')
  .pipe(trim())
  .subscribe(output => console.log(output.stdout.toString())); // Hello

throwIf(pattern: string | RegExp)

manually throw error if contents of stdout or stderr is matching supplied pattern

import {throwIf} from 'rxjs-shell';

exec('echo Hello').pipe(throwIf(/Hello/)).subscribe(); // ERROR

throwIfStdout(pattern: string | RegExp)

manually throw error if contents of stdout is matching supplied pattern

import {throwIfStdout} from 'rxjs-shell';

exec('echo Hello').pipe(throwIfStdout(/Hello/)).subscribe(); // ERROR
exec('>&2 echo Hello').pipe(throwIfStdout(/Hello/)).subscribe(); // OK

throwIfStderr(pattern: string | RegExp)

manually throw error if contents of stderr is matching supplied pattern

import {throwIfStderr} from 'rxjs-shell';

exec('echo Hello').pipe(throwIfStderr(/Hello/)).subscribe(); // OK
exec('>&2 echo Hello').pipe(throwIfStderr(/Hello/)).subscribe(); // ERR

execWithStdin(command)

executes a command with a string event as stdin input

of('Hello World')
  .pipe(execWithStdin('cat -'))
  .subscribe(output => {
    expect(String(output.stdout).trim()).to.equal('Hello World');
  });

Utility Methods

spawnEnd(spawnObservable: Observable) → Subject<{stdout: Buffer, stderr: Buffer}>

spawn emit each buffer from child process. if you want to connect other operator to this stream. use spawnEnd method.

import {spawn, spawnEnd} from 'rxjs-shell';

spawn('webpack', ['-p'])
  .pipe(outputChunk => {
    /* each child process's output buffer */
  })
  .subscribe();

spawnEnd(spawn('webpack', ['-p']))
  .pipe(webpackOutput => {
    /* do something */
  })
  .subscribe();

listenTerminating(fn: () => any)

invoke callbacks when one of signals that below is emitted.
- SIGINT
- SIGBREAK (for windows)

basically each operators are listen that. if user pressed ^C below stream is unsubscribe immediatly.

exec('curl ...')
  .pipe(concatMap(() => exec('curl ...')))
  .subscribe();

but if operators are not tied of one stream. whole process does not terminate. in this case. you can use listenTerminating.

import {exec, listenTerminating} from 'rxjs-shell';

// terminate process
listenTerminating(code => process.exit(code));
async () => {
  // user pressing ^C while curl is running
  await exec('curl ...').toPromise();

  // execute despite of pressing ^C. needs `listenTerminating`
  await exec('curl -X POST ...').toPromise();
};

isSpawnChunk(obj: any): obj is SpawnChunk

isExecOutput(obj: any): obj is ExecOutput

Error Handling

import {ShellError, spawn} from 'rxjs-shell';

spawn('git clone http://github.com/johnny-mh/rxjs-shell-operators')
  .pipe(tap(chunk => process.stdout.write(String(chunk.chunk))))
  .subscribe({
    catch(err) {
      if (!(err instanceof ShellError)) {
        throw err;
      }

      console.log(err.originError);
      console.log(err.stdout);
      console.log(err.stderr);
      console.log(err.toAnnotatedString()); // print annotated errors
    },
  });

FAQ

Operator does not throw script error

Some shell script doesn't completed with Non-Zero code. they just emitting error message to stderr or stdout 😢. If so. hard to throw ShellError because of err is null. You can use throwIf, throwIfStdout, throwIfStderr operator manually throwing specific scripts.

exec('sh a.sh')
  .pipe(concatMap(() => exec('sh b.sh').pipe(throwIf(/ERROR:/))))
  .subscribe();