掌握 JavaScript：利用高阶流释放函数响应式编程的力量

const { fromevent } = rxjs;
const { map, switchmap } = rxjs.operators;

const searchinput = document.getelementbyid('search-input');
const searchbutton = document.getelementbyid('search-button');

const searchstream = fromevent(searchbutton, 'click').pipe(
  map(() => searchinput.value),
  switchmap(query => fetchsearchresults(query))
);

searchstream.subscribe(results => {
  // display results
});

function fetchsearchresults(query) {
  // simulate api call
  return new promise(resolve => {
    settimeout(() => {
      resolve(`results for ${query}`);
    }, 1000);
  });
}

const { fromevent } = rxjs;
const { debouncetime, map } = rxjs.operators;

const editor = document.getelementbyid('editor');

const changestream = fromevent(editor, 'input').pipe(
  debouncetime(300),
  map(event => event.target.value)
);

changestream.subscribe(content => {
  sendtoserver(content);
});

function sendtoserver(content) {
  // simulated server send
  console.log('sending to server:', content);
}

const { behaviorsubject } = rxjs;
const { scan } = rxjs.operators;

const initialstate = { count: 0 };
const state$ = new behaviorsubject(initialstate);

const increment$ = new behaviorsubject(1);
const decrement$ = new behaviorsubject(-1);

const counter$ = state$.pipe(
  scan((state, change) => ({ count: state.count + change }), initialstate)
);

increment$.subscribe(state$);
decrement$.subscribe(state$);

counter$.subscribe(state => console.log(state.count));

// increment
increment$.next(1);
// decrement
decrement$.next(-1);

const { fromevent, merge } = rxjs;
const { map, takeuntil, switchmap } = rxjs.operators;

const draggable = document.getelementbyid('draggable');

const mousedown$ = fromevent(draggable, 'mousedown');
const mousemove$ = fromevent(document, 'mousemove');
const mouseup$ = fromevent(document, 'mouseup');

const drag$ = mousedown$.pipe(
  switchmap(start => {
    const startx = start.clientx - draggable.offsetleft;
    const starty = start.clienty - draggable.offsettop;

    return mousemove$.pipe(
      map(move => ({
        x: move.clientx - startx,
        y: move.clienty - starty
      })),
      takeuntil(mouseup$)
    );
  })
);

drag$.subscribe(pos => {
  draggable.style.left = `${pos.x}px`;
  draggable.style.top = `${pos.y}px`;
});

const { interval } = rxjs;
const { buffertime } = rxjs.operators;

const faststream$ = interval(10); // emits every 10ms

const bufferedstream$ = faststream$.pipe(
  buffertime(1000) // collect values for 1 second
);

bufferedstream$.subscribe(buffer => {
  console.log(`received ${buffer.length} values`);
});

const { observable } = rxjs;

function customoperator() {
  return (source$) => {
    return new observable(observer => {
      return source$.subscribe({
        next(value) {
          if (value % 2 === 0) {
            observer.next(value * 2);
          }
        },
        error(err) { observer.error(err); },
        complete() { observer.complete(); }
      });
    });
  };
}

const source$ = of(1, 2, 3, 4, 5);
const result$ = source$.pipe(customoperator());

result$.subscribe(x => console.log(x)); // outputs: 4, 8

const { of, throwerror } = rxjs;
const { mergemap, delay, retry } = rxjs.operators;

function fetchwithretry(url) {
  return of(url).pipe(
    mergemap(u => {
      // simulate a failing api call
      return math.random() < 0.5 ? throwerror('api error') : of(`response from ${u}`);
    }),
    retry({
      count: 3,
      delay: (error, retrycount) => {
        const delay = math.pow(2, retrycount) * 1000;
        console.log(`retrying in ${delay}ms`);
        return of(null).pipe(delay(delay));
      }
    })
  );
}

fetchwithretry('https://api.example.com')
  .subscribe(
    response => console.log(response),
    error => console.error('failed after 3 retries', error)
  );

const { combinelatest, behaviorsubject } = rxjs;

const userprofile$ = new behaviorsubject({ name: 'john' });
const userpreferences$ = new behaviorsubject({ theme: 'light' });
const currentroute$ = new behaviorsubject('/home');

const appstate$ = combinelatest([
  userprofile$,
  userpreferences$,
  currentroute$
]).pipe(
  map(([profile, preferences, route]) => ({
    profile,
    preferences,
    route
  }))
);

appstate$.subscribe(state => {
  console.log('app state updated:', state);
});

// update individual streams
userpreferences$.next({ theme: 'dark' });
currentroute$.next('/settings');

const { TestScheduler } = require('rxjs/testing');

describe('My Observable', () => {
  let testScheduler;

  beforeEach(() => {
    testScheduler = new TestScheduler((actual, expected) => {
      expect(actual).toEqual(expected);
    });
  });

  it('should filter even numbers', () => {
    testScheduler.run(({ cold, expectObservable }) => {
      const source$ = cold('a-b-c-d-e-|', { a: 1, b: 2, c: 3, d: 4, e: 5 });
      const expected = '---b---d-|';

      const result$ = source$.pipe(filter(x => x % 2 === 0));

      expectObservable(result$).toBe(expected, { b: 2, d: 4 });
    });
  });
});