bskydive

const auditProbe$ = item => { // функция-аргумент для передачи в оператор
	logAll('проверка: ' + item); // для отладки пишем полученное значение
	return interval(300).pipe(take(3)); // возвращаем наблюдатель. В данном случае - для имитации трёх значений. 
	//.pipe(take(X)) - хорошее правило для ограничения утечек памяти
}

const audit2$ = interval(102).pipe( // поток для отладки оператора
	take(10), // ограничиваем количество значений
	map(item => item * 102), // делаем значения человеко-понятными, выводим время их имитации в мсек, выбрали 102 вместо 100 чтобы не было случайных гонок асинхронных потоков(перестраховка)
	tap(logAll), // выводим сырые значения перед отправкой в недра исследуемого оператора

bskydive

public requireUsers$ = this.userSelectors.needUsers$
	.pipe(
		filter(needUsers => needUsers),
		tap(() => this.store.dispatch({ type: 'GET_USERS' })),
		switchMap(() => this.getUsers()),
		tap(users => this.store.dispatch({ type: 'RECEIVE_USERS', users })),
	)
	.share();

public users$ = this.muteFirst(

bskydive

ngOnInit() {
 this.users$ = this.userService.users$;
}

bskydive

public requireUsers$ = this.userSelectors.needUsers$
	.pipe(
		filter(needUsers => needUsers),
		tap(() => this.store.dispatch({ type: 'GET_USERS' })),
		switchMap(() => this.getUsers()),
		tap(users => this.store.dispatch({ type: 'RECEIVE_USERS', users }))
	)
	.finally(() => this.store.dispatch({ type: 'CANCEL_GET_USERS' }))
	.share();

bskydive

@Effect() actionX$ = this.updates$.pipe(
	ofType('ACTION_X'),
	map(toPayload),
	switchMap(payload => this.api.callApiX(payload)
		.pipe(
			map(data => ({ type: 'ACTION_X_SUCCESS', payload: data }))
		)
		.catch(err => Observable.of({ type: 'ACTION_X_FAIL', payload: err }))
	)
);

bskydive

@Effect()
getUsers$ = this.actions$.pipe(
	// отфильтровываем два типа действий
	ofType('GET_USERS', 'CANCEL_GET_USERS')
	// возвращаем в поток текущее действие селектора необходимости обновить пользователей
	withLatestFrom(this.userSelectors.needUsers$)
	// возвращаем в поток действие, когда есть необходимость обновить список пользователей
	filter(([action, needUsers]) => needUsers)
	// возвращаем в поток действие без флага needUsers
	map(([action, needUsers]) => action)

bskydive

// функция вызывается перед уничтожением компонента
ngOnDestroy() {
 // отправляем действие отмены запроса на сервер
 this.store.dispatch({type: "CANCEL_GET_USERS"})
}

bskydive

@Effect()
getUsers$ = this.actions$.pipe(
	// отфильтровываем действие получения списка пользователей
	ofType('GET_USERS'),
	// возвращает в поток текущее действие селектора необходимости обновить пользователей
	withLatestFrom(this.userSelectors.needUsers$),
	// возвращаем в поток действие, где есть необходимость обновить список пользователей
	filter(([action, needUsers]) => needUsers),
	// открывает новый поток с запросом на сервер
	switchMap(() => this.getUsers()),

bskydive

// функция вызывается при создании компонента
ngOnInit() {
  // отправляем действие получения списка пользователей
  this.store.dispatch({type: "GET_USERS"});
}

bskydive

export function favoriteItemsReducer(state = initialState, action: Action) {
  // выбираем реакцию по типу действия
  switch(action.type) {
    case 'REMOVE_FAVORITE_ITEM': 
    case 'DELETE_ITEM_SUCCESS': 
      // при удалении любимых записей и элементов действия одинаковые
      // берём из потока состояние массива любимых записей
      const itemId = action.payload; 
      // возвращаем состояние(объект) массива любимых записей без удалённой из дерева
      return state.filter(id => id !== itemId);