new_events_observable.js

new_events_observable.js

import * as Rx from 'rxjs/Rx';
import { API_ROOT } from 'src/constants';
import Axios, { AxiosResponse } from 'axios';
import * as moment from 'moment';

function createDatestamp() {
  return moment().utc().format('YYYY-MM-DDTHH:mm:ss');
}

let datestamp = createDatestamp();

type EventResponse = AxiosResponse<Linode.ManyResourceState<Linode.Event>>;

const initial$ = Rx.Observable
  .defer(() =>
    Rx.Observable
      .fromPromise(
        Axios.get(`${API_ROOT}/account/events`)
          .then(
            (response: EventResponse) => response.data.data,
          ),
  ));

/** Get events. */
const polling$ = Rx.Observable
  .interval(5000)
  .flatMap(() =>
    Rx.Observable
      .fromPromise(
        Axios.get(
          `${API_ROOT}/account/events`,
          { headers: { 'X-Filter': JSON.stringify({ created: { '+gt': datestamp } }) } })
          .then((response: EventResponse) => response.data.data),
      ),
  );

linodeEvents$ = Rx.Observable();

const stream$: Rx.Observable<Linode.Event[]> = Rx.Observable
  .merge(initial$, polling$)
  .scan(
    (_, value: Linode.Event[]) => {
      if (value[0]) {
        datestamp = value[0].created;
      }
      
      value.map(lEvent => linodeEvents$.next(lEvent))
      
      return null; /* nothing consumes this */
    },
    null,
  );
  
linodeEvents$.publish();
  
/** Example of how we'd get the events for the dropdown. */
exp$.subscribe(
    (event: Linode.Event) => console.log('Event: ', event),
);

/** Example of how we would get the badge count. */
exp$
  .map(event => /* use this event to update a leaky bucket in local state*/)

exp$.connect();