import {Injectable} from '@angular/core';
import {HttpClient} from '@angular/common/http';
import {EMPTY, Observable} from 'rxjs';
import {catchError, switchMap, tap} from 'rxjs/operators';
import {UserState} from './user.state';
​
// Create a separate service for Streams, it helps avoid cyclic dependencies.
// It'll do all the heavy lifting, it'll make http requests and
// dispatch the results to the userDataSystem.
​
@Injectable() // we'll instantiate it in user.module.ts so no "provideIn"
export class UserStreams {
​
  // create a Stream using the userSystem and couple it with an http request
  userDataStream = this.userState.userSystem.createStream(
    (
      queryUnit, dataUnit, errorUnit
    ) => {
      // listen to the future values of queryUnit
      return queryUnit.future$.pipe(
      // you can also do queryUnit.pipe, but that'll start immediately,
      // without even dispatching anything to queryUnit
  
        // switch the stream to http request Observable
        // to create a new request
        switchMap(userId => this.httpClient
          .get('https://example.com/username' + userId)
          .pipe(
            // dispatch the returned data to dataUnit
            tap(data => dataUnit.dispatch(data)),
    
            // catch the error
            catchError(err => {
    
              // dispatch the error to errorUnit
              errorUnit.dispatch(err);
    
              // don't rethrow, to keep the stream alive
              return EMPTY;
            })
          )
        ));
  });
​
  constructor(
    private httpClient: HttpClient, 
    private userState: UserState
  ) {}
}

import {Component} from '@angular/core';
import {UserState} from './user.state';
​
/* 
  An Angular component.
  It will trigger the API request and show the response 
  along with loading-state.
*/
​
@Component({
  selector: 'app-user',
  templateUrl: 'user.component.html'
})
export class UserComponent {
  userId: number;
​
  constructor(public userState: UserState) {}
​
  loadData() {
    this.userState.userQuery.dispatch(this.userId);
  }
​
  reloadData() {
    this.userState.userQuery.replay();
  }
}

<!--> 
  The template for Angular component.
</-->
​
<h1>Load data from a REST API using AsyncSystem.</h1>
​
<input type="number" [(ngModel)]="userId">
​
<button (click)="loadData()">Load Data</button>
<button (click)="reloadData()">Reload Data</button>
​
<p *ngIf="userState.userPending | async">
  Loading...
</p>
​
<p *ngIf="userState.userData | async as name">
  My name is: {{name}}
</p>
​
<p *ngIf="userState.userError | async as error">
  Error: {{error.message}}
</p>

import {NgModule} from '@angular/core';
import {CommonModule} from '@angular/common';
import {UserStreams} from './user.streams';
import {UserComponent} from './user.component';
​
// here we are creating a Feature Module, so we're making the UserStreams part of
// this module only, so they start working when this Module is used.
// But if we had more generic Streams-service we can do the same
// in the AppModule so that the Streams start when the app starts.
​
@NgModule({
  declarations: [UserComponent],
  imports: [CommonModule],
  providers: [
    // make the UserStreams part of dependecy injection
    UserStreams,
    // instantiate the UserStreams service using APP_INITILIZER, or
    // we can do it in old fashioned way by putting in the contructor of AppModule
    {
      provide: APP_INITIALIZER,
      useFactory: () => () => {},
      deps: [UserStreams], // we can put as many Streams services here as we want
      multi: true,
    },
  ]
})
export class UserModule {}

// Create a separate file for Streams, it helps avoid cyclic dependencies.
// It'll do all the heavy lifting, it'll make http requests and
// dispatch the results to the userDataSystem.
​
import {EMPTY, Observable} from 'rxjs';
import {catchError, switchMap, tap} from 'rxjs/operators';
import {queryUnit, dataUnit, errorUnit} from './user.state';
​
// a function to fetch data and disptch the response appropriately
async function fetchAndShareData(query) {
  try {
    // fetch data using fetch API
    const response = await fetch('https://xyz.com/u/' + query.userId);
    // extract the JSON data
    const data = await response.json();
​
    // dispatch data to the dataUnit, it also sets the pendingUnit's value to false
    // and it also clears the errorUnit's value
    dataUnit.dispatch(data);
  } catch (err) {
    // dispatch error to errorUnit, it also sets the pendingUnit's value to false
    errorUnit.dispatch(err);
  }
}
​
// setup the stream by observing query values, that triggers fetchAndShareData
// whenever a value is dispatched to queryUnit
queryUnit.subscribe(query => fetchAndShareData(query));

/* 
  A React component.
  It will trigger the API request and show the response 
  along with loading-state.
*/
function App() {
  const [userId, setUserId] = useState('');
  
  const data = useObservable(userDataUnit);
  const isPending = useObservable(userPendingUnit);
  const error = useObservable(userErrorUnit);
​
  const loadData = () => userQueryUnit.dispatch(userId);
  const reloadData = () => userQueryUnit.replay();
​
  return (
    <React.Fragment>
      <h1>Load data from a REST API using AsyncSystem.</h1>
      {userId}
​
      <input type="number" onChange={e => setUserId(e.target.value)}/>
​
      {isPending && <p>Loading...</p>}
​
      {data && <p>My name is: {data}</p>}
​
      {error && <p>Error: {error}</p>}
​
      <button onClick={loadData}>Load Data</button>
      <button onClick={reloadData}>Reload</button>
    </React.Fragment>
  );
}