---
# System prepended metadata

title: ReactiveJS

---

# ReactiveJS

## Functional Reactive Programming (FRP)
### 函數式程式

* 聲明式
* 純函數
* 資料不可變

#### 命令式、聲明式
```javascript=
命令式
function double(arr) {
    const results = []
    for (let i = 0; i < arr.length; i++) {
        results.push(arr[i] * 2) 
    }
    return results
}

function addOne(arr) {
    const results = []
    for (let i = 0; i < arr.length; i++) {
        results.push(arr[i] + 1) }
    return results
}
```

```javascript=
聲明式
function double(arr) {
    return arr.map(function(item) {return item * 2});
}

function addOne(arr) {
    return arr.map(function(item) {return item + 1}); 
}
```

```javascript=
const double = arr => arr.map(item => item * 2); 
const addOne = arr => arr.map(item => item + 1);
```



### 響應式程式


EXCEL

![](https://i.imgur.com/TcxnOUT.png)

### Reactive Extension
![](https://i.imgur.com/793FQJa.png)

## 拉(PULL)、推(PUSH)
PULL和PUSH描述了數據生產者如何與數據使用者溝通。

PULL -> 數據的使用者決定了何時需要取得資料。
PUSH -> 數據的生產者決定了何時需要傳送資料。

## Design pattern
Observer Pattern
![](https://i.imgur.com/dY4CP4u.png)
https://www.tutorialspoint.com/design_pattern/observer_pattern.htm
Iterator Pattern
![](https://i.imgur.com/SzwoLtb.png)
https://www.tutorialspoint.com/design_pattern/iterator_pattern.htm

## 核心概念
1. Observable - 可觀察的物件
2. Observer - 觀察者物件
3. Subscription - 訂閱物件
4. Operators - 運算子
5. Subject - 廣播資料傳個多個 Observer
6. Schedulers - 控制事件併發狀況

## Observer
觀察者物件
```javascript=
observable$.subscribe({
    next: () => {},
    error: () => {},
    complete: () => {},
})
```

## Subscription
訂閱物件

```javascript=
const subscription = observable$.subscribe({
    next: () => {},
    error: () => {},
    complete: () => {},
})

subscription.unsubscribe();
```

## Operators
運算子

```javascript=
const subscription = observable$.pipe(
    map((e) => return e + 'test')
)
.subscribe({
    next: () => {},
    error: () => {},
    complete: () => {},
})

```

## Subject
廣播資料傳個多個 Observer

```javascript=
import { Subject, from } from 'rxjs';
 
const subject = new Subject<number>();
 
subject.subscribe({
  next: (v) => console.log(`observerA: ${v}`)
});
subject.subscribe({
  next: (v) => console.log(`observerB: ${v}`)
});
 
const observable = from([1, 2, 3]);
 
observable.subscribe(subject); // You can subscribe providing a Subject
 
// Logs:
// observerA: 1
// observerB: 1
// observerA: 2
// observerB: 2
// observerA: 3
// observerB: 3
```

## Observable

### 基本Observable
```javascript=
const observable$ = new Observable(observer => {
    observer.next(1);
    observer.next(2);
    observer.next(3);
    observer.error('error occur');
    observer.complete();
});

const sub = observable$.subscribe({
    next: data => console.log(data),
    complete : () => console.log('complete'),
    error: (error) => console.log(error)
});
```
* 練習1. 實現顯示持續計數增加的Observable，並在十秒後完成。

```javascript=
const observable$ = new Observable(observer => {
})
const sub = observable$.subscribe({
    next: data => console.log(data),
    complete : () => console.log('complete'),
    error: (error) => console.log(error)
})
/*
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    complete
*/
```

### 基本Observable - unsubscribe()
```javascript=
const observable$ = new Observable(observer => {
    observer.next(1);
    observer.next(2);
    observer.next(3);
    return () => {
        console.log('unsubscribe');
    }
});

const sub = observable$.subscribe({
    next: data => console.log(data),
    complete : () => console.log('complete'),
    error: (error) => console.log(error)
})

sub.unsubscribe();
```

* 練習2. 實現顯示持續計數增加的Observable，並在五秒後停止訂閱。

```javascript=
const observable$ = new Observable(observer => {
});

const sub = observable$.subscribe({
    next: data => console.log(data),
    complete : () => console.log('complete'),
    error: (error) => console.log(error)
})
setTimeout(() => sub.unsubscribe(), 5000);
```

### 遍歷 Observable
```javascript=
const arr = [1,2,3,4];

function getArrObservable(arr) {
    return new Observable(observer => {
        arr.forEach(function(element) {
            observer.next(element);
        });
        observer.complete();
    });
})

getArrObservable(arr).subscribe({
    next: data => console.log(data),
    error: error => console.log(error),
    complete: () => console.log('complete')
})

```

### Promise Observable

```javascript=
function getPromiseToObservable(myPromise) {
    return new Observable(observer => {
        myPromise
            .then(data => observer.next(data))
            .catch(err => observer.error(err))
            .finally(() => observer.complete())
    });
}

const myPromise = new Promise((resolve, reject) => {
    setTimeout(() => {
        resolve('Hello! World!');
    }, 3000)
})

getPromiseToObservable(myPromise).subscribe({
    next: data => console.log(data),
    complete : () => console.log('complete'),
    error: (error) => console.log(error)
})
```

### HttpRequest Observable
```javascript=
function getObservable(url) {
    return new Observable(observer => {
        const xhr = new XMLHttpRequest();
        xhr.addEventListener('load', () => {
            if (xhr.readyState === 4 && xhr.status === 200) {
                observer.next(JSON.parse(xhr.responseText));
                observer.complete();
            }
        });

        xhr.open('GET', url);
        xhr.send();
        return () => xhr.abort()
    })
}

const url = 'https://jsonplaceholder.typicode.com/todos';
const sub = getObservable(url).subscribe({
    next: data => console.log(data),
    complete : () => console.log('complete'),
    error: (error) => console.log(error)
});
sub.unsubscribe();
```
* 練習3. 建立一個可產生倒數N秒Observable的function。
```javascript=
function getCountTimeObservable(max) {

}

getCountTimeObservable(10).subscribe(data => console.log(data));
```

## promise array fromEvent http

* fromEvent
```javascript=
import { fromEvent } from 'rxjs';

const clicks = fromEvent(document, 'click');
clicks.subscribe(x => console.log(x));
```

* array
```javascript=
const arr = [1,2,3,4,5,6,7]

from(arr)
    .pipe(
      filter(e => e > 3),
      toArray()
    )
    .subscribe(e => console.log(e));
```

* Promise
```javascript=
const promise$ = new Promise((resolve, reject) => {
    setTimeout(() => {
      resolve('success');
    }, 3000);
});

from(promise$).subscribe(e => console.log(e));
```

* Http
```javascript=
this.http.get('https://jsonplaceholder.typicode.com/todos').subscribe({
      next: data => console.log(data),
      complete: () => console.log('complete'),
      error: error => console.log(error)
});
```

## Operators
### Operators - merge, concat, tap, filter, map, take
創建類（creation）
轉化類（transformation）
過濾類（filtering）
合併類（combination）
多播類（multicasting）
錯誤處理類（error Handling）
輔助工具類（utility）
條件分支類型（conditional&boolean）
數學和合計類（mathmatical&aggregate）

* of
將引數轉換為Observable
```
of('Hello').subscribe(res => console.log(res));
```

* from
從數列、Promise、iterator等等物件建立一個Observable
```
from([1, 2, 3, 4, 5]).subscribe(res => console.log(res));
```

* map
其實與javascript相同，是資料的映射。
```
const source$ = of(1,2,3,4,5);
source$
    .pipe(
        map(element => element + 1)
    )
    .subscribe(element => console.log)
```

* filter
其實與javascript相同，是資料的過濾。
```
const source$ = of(1,2,3,4,5,6);

source$
  .pipe(
    filter(element => element > 3)
  )
  .subscribe(element => console.log(element)); 
```

* tap
監聽，並不影響上游下游資料。
```
const source$ = of(1,2,3,4,5,6);

source$
  .pipe(
    tap(e => console.log(e))
  )
  .subscribe(element => console.log(element)); 
```

* take
就是取得，從上游Observable拿資料，拿夠了就會complete，拿多少才夠就由take決定。且只有取得足夠的數量後，就會立刻complete。
```
const timer$ = interval(1000);

timer$
    .pipe(take(1))
    .subscribe({
      next: (data) => console.log(data),
      complete: () => console.log('complete')
    });
```


* concat
與javascript的array concat相似，可以合併兩個或多個Observable。
```
const arr1$ = of(1,2,3,4,5);
const arr2$ = of(6,7,8,9,10);
const arr3$ = concat(arr1$, arr2$);
arr3$.subscribe(element => console.log(element));
```

* merge
merge與concat用法很相似，但功能很不一樣，merge會第一時間訂閱所有上游的Observable，採先到先得的策略。
```
const timer1$ = timer(0, 1000);
const timer2$ = timer(500, 1000);
const merge$ = merge(timer1$, timer2$);
merge$.subscribe(data => console.log(data))
```

* 練習4. 將陣列`[1,2,3,4,5]`每一個乘三並留下大於五的數。
```javascript=
const arr = [1,2,3,4,5];
```

### High Order Operators - switchmap, concatmap, mergemap


處理巢狀Obersable
```javascript=
let postApiUrl = 'https://jsonplaceholder.typicode.com/posts/1';
let commentApiUrl = 'https://jsonplaceholder.typicode.com/comments';

this.http.get(postApiUrl).subscribe((p: any) => {
  this.http.get(`${commentApiUrl}/${p.id}`).subscribe(c => {
    console.log(c);
  })
});
```

```javascript=
const timer1$ = timer(2000).pipe(tap(() => console.log('2000OK')));
const timer2$ = timer(5000).pipe(tap(() => console.log('5000OK')));
const timer3$ = timer(1000).pipe(tap(() => console.log('1000OK')));
const arr = [timer1$, timer2$, timer3$];
```

* 取得每一個post的comments
```javascript=
let postApiUrl = 'https://jsonplaceholder.typicode.com/posts';
let commentApiUrl = 'https://jsonplaceholder.typicode.com/comments';

```

## Subject
![](https://i.imgur.com/zZ93fFQ.png)

```javascript=
import { Subject } from 'rxjs';
 
const subject = new Subject<number>();
 
subject.subscribe({
  next: (v) => console.log(`observerA: ${v}`)
});

subject.subscribe({
  next: (v) => console.log(`observerB: ${v}`)
});
 
subject.next(1);
subject.next(2);
 
// Logs:
// observerA: 1
// observerB: 1
// observerA: 2
// observerB: 2
```

比較(單一訊號多個訂閱)
```javascript=
const timer1 = interval(2000).pipe(take(5));

timer1.subscribe(x => console.log(x));

setTimeout(() => {
  timer1.subscribe(x => console.log(`%c三秒後 ${x}`, 'color: red'));
}, 3000);
```

subject
```javascript=
const timer$ = interval(1000).pipe(take(5));

const subject = new Subject();
timer$.subscribe(subject);

subject.subscribe(x => console.log(x));

setTimeout(() => {
  subject.subscribe(x => console.log(`%c三秒後 ${x}`, 'color: red'));
}, 3000);
```