https://www.youtube.com/watch?v=Z76QlSpYcck&list=PLOETEcp3DkCpimylVKTDe968yNmNIajlR&index=16
Data Composition with RxJS | Deborah Kurata

我這一篇算是
Day25_[S05E07] CaseStudy: RxJS真實案例展示
的延申閱讀

在[S05E07]，jiaming大大分享怎麼從多個Web API傳回來的observable
用RxJS來重組成我們需要的資料

我覺得跟這一篇有關
Data Composition with RxJS | Deborah Kurata

如果你有興趣，可以直接看影片，有英文字幕，
這一集相較custom operator，比較沒那麼硬，
而且將HttpClient跟後端要到資料後的組合，非常實用!!

投影片
http://bit.ly/deborahk-ngconf2019
原始碼
https://github.com/DeborahK/Angular-DD
Twitter:
deborahkurata

目標

Declarative approach to

• Collecting
• Composing(combinding)
• Caching
  using RxJS streams with NO subscriptions
  (沒有subscribe，就沒有unsubscrib)

Collect Data

一般的http get

1. 假設有一個Service

@Injectable({ providedIn:'root' })
export class ProductService{
    private productsUrl='api/products';
    constructor(private http: HttpClient){}
    
    getProducts():Observable<Product[]>{
        return this.http.get<Product[]>(this.productsUrl)
            .pipe(
                tap(console.log),
                catchError(this.handleError)
            );
    }
}

2. 一個Component使用service

export class ProductListComponent implements OnInit{
    products$=this.productService.products$
        .pipe(
            catchError(error=>{
                this.errorMessage=error;
                return of(null); // 是不是類似 return EMPTY ?
            })
        );
    constructor(private productService: ProductService){}
}

3. Template

<div *ngIf="products$ | async as products">
                        ^^^^^ 會自動訂閱及取消訂閱
    <button type='button'
        *ngFor='let product of products'>
        {{ product.productName }} ({{ product.category }})
    </button>
</div>

ChangeDetectionStrategy.OnPush

@Component({
    selector: 'pm-product-list',
    templateUrl: './product-list.component.html',
    changeDetection: ChangeDetectionStrategy.OnPush
                                             ^^^^^^
})

OnPush 改善效能(縮小change detection cycles)
只有在下列情況下會檢查

• @Input property changes
• An event emits
• A bound observable emits

Composing

1. 定義一個ProductCategory Service

@Injectable({ providedIn:'root' })
export class ProductCategoryService{
    private productCategoriesUrl='api/productCategories';
    constructor(private http: HttpClient){}
    
    productCategories$ = this.http.get<ProductCategory[]>(this.productCategoriesUrl)
            .pipe(
                tap(console.log),
                catchError(this.handleError)
            );
    }
}

• RxJS Creation Function: combineLatest()

• Uses the latest emitted values of each of its input streams

• Waits for each input stream to emit at least once before it starts emitting

• Emits an array of values, in order, one for each input stream

productsWithCategory$ = combineLatest(
                        ^^^^^^^^^^^^^
    this.products$,
    this.ProductCategoryService.productCategories$
) // 類似 [ Products[], ProductCategories[] ]
.pipe(   vvvvvvvvvvvvvvvvvvvvvvvv Array destructuring
    map( ([products, categories]) =>
                 VVV Array map method
        products.map(
            p=>
                (
                {   VVV Spread Operator copies over the properties
                    ...p,
                    category: categories.find(c=>
                        p.categoryId === c.id).name
                } as Product
                )
        )
    )
)

以上是類似
Day25_[S05E07] CaseStudy: RxJS真實案例展示
資料的重組

接下來是其他的資料流重組

Reacting to Action

1. Create an action stream

2. Combine action and data streams

3. Emit a value every time an action occurs

4. Creating an Action Stream

private productSelectedAction = new Subject<number>();
productSelectedAction$=this.productSelectedAction.asObservable();
                                                  ^^^^^^^^^^^^^^
                                // Observable
                                // Observer:next,error,complete

• Use Subject/BehaviorSubject(建立時可以傳資料當初使值)
• Special type of Observable
• Multicast

2. Combining Data and Action Streams
   merge the streams

selectedProduct$=combineLatest(
    this.productSelectedAction, // action stream (使用者選了某product)
    this.productsWithCategory$
).pipe(
    map( ([selectedProductId, products] ) =>
        products.find(product => product.id === selectedProductId)
                                                ^^^^^^^^^^^^^^^^^
    )
);

3. Emitting a Value on Action
   change detection(更新UI)
   make sure those emitting actions

// 當使用者選擇product的時候
onSelected(productId: number): void{
    this.productService.changeSelectedProduct(productId);
}                       ^^^^^^^^^^^^^^^^^^^^^ 告知service，productId變了
// service來操作observable$
changeSelectedProduct(selectedProductId: number | null):void{
    // emits that next productId
    this.productSelectedAction.next(selectedProductId);
                                     ^^^^^^^^^^^^^^^^ emitting a value on action
}

Emitting Re-executes the Pipeline

// marble diagram
data$ :  ----[ {saw},{rake},{axe} ]----
action$: --1-------------------------------14-
combineLatest(data$,action$) // 抓2個最新的，回傳新的Observable
         ----[ [{saw},{rake},{axe}] , 1 ]--[ [{saw},{rake},{axe}] , 14 ]

Caching

Cache the Array?
只能透過service分享arrar?

@Injectable({ providedIn: 'root' })
export class ProductService{
    private productsUrl='api/products';
    products: Product[];
    
    products$=this.http.get<Product[]>(this.productsUrl)
        .pipe(
            map(data=>this.products=data),
            catchError(this.handleError)
        );
    constructor(private http: HttpClient){}
}

RxJS operator: shareReplay

• Shares an Observable with all subscribers
• Replays the specified number of entries upon subscription
• Remains active, even if there are no current subscribers

    products$=this.http.get<Product[]>(this.productsUrl)
        .pipe(
            tap(console.log),
            shareReplay(),
            catchError(this.handleError)
        );

Observable All the Things

當Observable的streams改變時，是用推動 UI資料更新
Component

pageTitle$=this.product$
    .pipe(
        map((p:Product)=>
            p ? `Product Detail for: ${p.productName}` : null)
    );

// combine所有的streams
vm$=combineLatest(
    [this.product$,this.productSuppliers$,this.pageTitle$])
    .pipe(                  vvvvvvvvv filter掉還沒選的case
        filter(([product])=>!!product),
        // Array destructuring，轉成一個Object
        map( ([product,productSuppliers,pageTitle]) =>
            ({ product,productSuppliers,pageTitle }) )
    );

Template(使用)

<div *ngIf="vm$ | async as vm">
<div>{{vm.pageTitle}}</div>
<div>{{vm.product.productName}}</div>
...
<tr *ngFor="let supplier of vm.productSuppliers">
...

Declarative, Reactive Streams

困難點

• Picking the right operator is not always easy
• Difficult to debug Observables
  優點
• Composable streams
• Leverages power of RxJS operators
• Sharing Observables is easy(cache)
• Effectively react to actions
• Less code than NgRx
• OnPush for improved performance