Creating Operator

1-multiplyBy.js

const dom = $('#operator');
const foo = Rx.Observable.of(1, 2, 3, 4);

function multiplyByTen() {
  const sourceObs = this;
  const currentObs = Rx.Observable.create((observer) => {
    sourceObs.subscribe(
      (data) => observer.next(data * 10),
      (err) => observer.error(err),
      () => observer.complete()
    );
  });
  
  return currentObs;
}

Rx.Observable.prototype.multiplyByTen = multiplyByTen;

const bar = foo.multiplyByTen();

bar.subscribe(
  (data) => dom.append('<li>' + data + '</li>'),
  (err) => dom.append(err),
  () => dom.append('<li>done</li>')
);

10-scan.js

const dom = $('#operator');
const foo = Rx.Observable.of('h', 'e', 'l', 'l', 'o');
const bar = Rx.Observable.interval(600).take(5);

const combined = foo.zip(bar, (x,y) => x).scan((acc ,x) => acc+x , '');

/**
 * Marble Diagram
 * 
  hello|
  -----0-----1-----2-----3-----4|
      zip((x, y) => x)
  -----h-----e-----l-----l-----o|
      scan((acc, x) => acc+x, '')
  ---- h-----he----hel---hell--hello|

 */

combined.subscribe(
  (data) => dom.append('<li>' + data + '</li>'),
  (err) => dom.append(err),
  () => dom.append('<li>done</li>')
);

11-buffer.js

const dom = $('#operator');
const foo = Rx.Observable.of('h', 'e', 'l', 'l', 'o')
      .zip(Rx.Observable.interval(600).take(5), (x, y) => x);
const bar = Rx.Observable.interval(900).take(3);


/**
 * Marble Diagram  first n-th + second n-th
 * 
  hello|
  -----0-----1-----2-----3-----4|
      zip((x, y) => x)
  -----h-----e-----l-----l-----o|
      bufferCount(2)
  -----------he----------ll-----

 */
// const result = foo.bufferCount(2);


/**
 * 
   -----h-----e-----l-----l-----o|
   --------0--------1--------2|
      buffer(bar)
   --------h--------el-------l|
   
 */
const result = foo.buffer(bar);

result.subscribe(
  (data) => dom.append('<li>' + data + '</li>'),
  (err) => dom.append(err),
  () => dom.append('<li>done</li>')
);

12-delay.js

const dom = $('#operator');
const foo = Rx.Observable.interval(500).take(5);


/**
 * Marble Diagram  
 * 
  --0--1--2--3--4|
    delayWhen(fn)
  --0--1--2---3----4|

 */
const result = foo.delayWhen(x =>
  Rx.Observable.interval( x* x* 100).take(1)
);

result.subscribe(
  (data) => dom.append('<li>' + data + '</li>'),
  (err) => dom.append(err),
  () => dom.append('<li>done</li>')
);

13-debounce.js

const dom = $('#operator');
const foo = Rx.Observable.interval(100).take(5);


/**
 * Marble Diagram  
 * 
  --0--1--2--3--4|
    debounceTime(1000)
             ------------4|  debounceTime(1000);
  ---0--1--2--3--4| debounceTime(50);
 */
const result = foo.debounceTime(50);

// const result = foo.debounce(() => Rx.Observable.interval(1000).take(1));

result.subscribe(
  (data) => dom.append('<li>' + data + '</li>'),
  (err) => dom.append(err),
  () => dom.append('<li>done</li>')
);

14-throttle.js

const dom = $('#operator');
const foo = Rx.Observable.interval(500).take(5);


/**
 * Marble Diagram  
 * 
  --0--1--2--3--4|
    debounceTime(1000) // wait for silence, then emits
  --------------4|  
    
  --0--1--2--3--4|
    throttleTime(1000) // first emits, then causes silence
  --0-----2-----4|
 */
const result = foo.debounceTime(700);

result.subscribe(
  (data) => dom.append('<li>' + data + '</li>'),
  (err) => dom.append(err),
  () => dom.append('<li>done</li>')
);

15-distinct.js

const dom = $('#operator');
const foo = Rx.Observable.of('h', 'e', 'L', 'l', 'o')
      .zip(Rx.Observable.interval(600).take(5), (x, y) => x);


/**
 * Marble Diagram 
 * 
  hello|
  -----0-----1-----2-----3-----4|
      zip((x, y) => x)
  -----h-----e-----l-----l-----o|

 */
const result = foo.distinct();

//const result = foo.distinct((x, y) => x.toLowerCase() === y.toLowerCase());


/**
 * 
 ---a---b--aaaab|
  distinctUntilChanged
 ---a---b--a---b|
 */
result.subscribe(
  (data) => dom.append('<li>' + data + '</li>'),
  (err) => dom.append(err),
  () => dom.append('<li>done</li>')
);

16-catch.js

const dom = $('#operator');

const foo = Rx.Observable.interval(500)
  .zip(Rx.Observable.of('a', 'b', 'c', 'd', 10), (x, y) => y);

const bar = foo.map(x => x.toUpperCase());

/**
 * Marble Diagram 
 * 
  -----a-----b-----c-----d-----2| foo
      map(toUppserCase)
  -----A-----B-----C-----D-----X  bar
      catch
  -----A-----B-----C-----D-----Z|

 */
 
const result = bar.catch(error => Rx.Observable.of('Z'));

// retry value again
// const result = bar.catch((error, outputObs) => outputObs);

result.subscribe(
  (data) => dom.append('<li>' + data + '</li>'),
  (err) => dom.append(err),
  () => dom.append('<li>done</li>')
);

17-retry.js

const dom = $('#operator');

const foo = Rx.Observable.interval(500)
  .zip(Rx.Observable.of('a', 'b', 'c', 'd', 10), (x, y) => y);

const bar = foo.map(x => x.toUpperCase());

/**
 * Marble Diagram 
 * 
  -----a-----b-----c-----d-----2| foo
      map(toUppserCase)
  -----A-----B-----C-----D-----X  bar
      retry
  -----A-----B-----C-----D------------------A-----B-----C-----D------------------A-----B-----C-----D|

 */
 
// const result = bar.retry(2);

const result = bar.retryWhen(errorObs => errorObs.delay(1000));

result.subscribe(
  (data) => dom.append('<li>' + data + '</li>'),
  (err) => dom.append(err),
  () => dom.append('<li>done</li>')
);

18-repeat.js

const dom = $('#operator');

const foo = Rx.Observable.interval(500)
  .zip(Rx.Observable.of('a', 'b', 'c', 'd'), (x, y) => y);

const bar = foo.map(x => x.toUpperCase());

/**
 * Marble Diagram 
 * 
  -----a-----b-----c-----d| foo
      map(toUppserCase)
  -----A-----B-----C-----D|  bar
      retry
  -----A-----B-----C-----D-----A-----B-----C-----D|

 */
 
const result = bar.repeat(2);

result.subscribe(
  (data) => dom.append('<li>' + data + '</li>'),
  (err) => dom.append(err),
  () => dom.append('<li>done</li>')
);

2-map.js

const dom = $('#operator');
const foo = Rx.Observable.of(1, 2, 3, 4);

function calc(transformFn) {
  const sourceObs = this;
  const currentObs = Rx.Observable.create((observer) => {
    sourceObs.subscribe(
      (data) => observer.next(transformFn(data)),
      (err) => observer.error(err),
      () => observer.complete()
    );
  });
  
  return currentObs;
}

Rx.Observable.prototype.calc = calc;

const bar = foo.calc(x => x + 3);

bar.subscribe(
  (data) => dom.append('<li>' + data + '</li>'),
  (err) => dom.append(err),
  () => dom.append('<li>done</li>')
);

3-do.js

const dom = $('#operator');
const foo = Rx.Observable.of(1, 2, 3, 4);


const bar = foo.do(x => {
  console.log(x);
});

// try before / after do between map using chaining operator


bar.subscribe(
  (data) => dom.append('<li>' + data + '</li>'),
  (err) => dom.append(err),
  () => dom.append('<li>done</li>')
);

4-take.js

const dom = $('#operator');
const foo = Rx.Observable.of(1, 2, 3, 4);


const bar = foo.skip(2);

// takeLast, first, last, skip


bar.subscribe(
  (data) => dom.append('<li>' + data + '</li>'),
  (err) => dom.append(err),
  () => dom.append('<li>done</li>')
);

5-concat.js

const dom = $('#operator');
const foo = Rx.Observable.interval(1000).take(4);
const more = Rx.Observable.of(5, 6, 7, 8);

const bar = foo.concat(more);

// const prefix = Rx.Observable.of('a');
// prefix.concat(foo) === foo.?('a')

bar.subscribe(
  (data) => dom.append('<li>' + data + '</li>'),
  (err) => dom.append(err),
  () => dom.append('<li>done</li>')
);

6-merge.js

const dom = $('#operator');
const foo = Rx.Observable.interval(500).take(4);
const bar = Rx.Observable.interval(300).map(x => 'a').take(5);

const merge = foo.merge(bar);

/**
 * Marble Diagram 
 * 
 ----0----1----2----3|
 --0--1--2--3--4|
    merge OR
 --0-01--21-3--24---3|
 */

merge.subscribe(
  (data) => dom.append('<li>' + data + '</li>'),
  (err) => dom.append(err),
  () => dom.append('<li>done</li>')
);

7-combineLatest.js

const dom = $('#operator');
const foo = Rx.Observable.interval(500).take(4);
const bar = Rx.Observable.interval(300).take(5);

const combined = foo.combineLatest(bar, (x,y) => x+y);

/**
 * Marble Diagram 
 * 
 ----0----1----2----3|
 --0--1--2--3--4|
    combineLatest(fn)
 ----01--23-4--56---7|
 
    merge: OR
    combineLatest: AND
 */

combined.subscribe(
  (data) => dom.append('<li>' + data + '</li>'),
  (err) => dom.append(err),
  () => dom.append('<li>done</li>')
);

8-withLatestFrom.js

const dom = $('#operator');
const foo = Rx.Observable.interval(500).take(5)
  .zip(Rx.Observable.of('H', 'e', 'l', 'l', 'o'), (_, c) => c);
const bar = Rx.Observable.interval(300).take(7)
  .zip(Rx.Observable.of(0, 1, 0, 1, 0, 1, 0), (_, x) => x);

const mapped = foo.withLatestFrom(bar, (c, n) => n===1 ? 
  c.toUpperCase(): c.toLowerCase());

/**
 * Marble Diagram 
 * 
  ----H----e----l----l----o|
  --0--1--0--1--0--1--0|
    withLatestFrom
  ----h----e----L----L----o|

 */

merge.subscribe(
  (data) => dom.append('<li>' + data + '</li>'),
  (err) => dom.append(err),
  () => dom.append('<li>done</li>')
);

9-zip.js

const dom = $('#operator');
const foo = Rx.Observable.interval(500).take(5);
const bar = Rx.Observable.interval(400).take(4);

const combined = foo.zip(bar, (x,y) => x+y);

/**
 * Marble Diagram  first n-th + second n-th
 * 
  ----0----1----2----3---4|
  ---0---1---2---3|
      zip
  ----0----2----4----6|

 */
 
 // let's try print marble-diagram of hello per 1 second using zip
 // const foo = Rx.Observable.of('h', 'l', 'l', 'o');
 // foo.zip(bar, (x, y) => x);

combined.subscribe(
  (data) => dom.append('<li>' + data + '</li>'),
  (err) => dom.append(err),
  () => dom.append('<li>done</li>')
);

index.html

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <title>GistRun</title>
  <script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.4.0/Rx.js"></script>
  <script src="https://code.jquery.com/jquery-2.2.4.min.js"></script>
  <link rel="stylesheet" href="styles.css">
</head>
<body>
  <h1>Creating Operator</h1>
  <div id="operator"></div>
  
  <script src="18-repeat.js"></script>
</body>
</html>

styles.css

/* todo: add styles */