How to use rx-lite - 10 common examples

var func1: (a: string, cb: (result: number) => void) => number = () => 0;
		obsNum = Rx.Observable.fromCallback(func1)("");
		obsNum = Rx.Observable.fromCallback(func1, {})("");
		obsNum = Rx.Observable.fromCallback(func1, {}, (results: number[]) => results[0])("");

		// 2 arguments
		var func2: (a: number, b: string, cb: (result: string) => number) => Date = () => new Date();
		obsStr = Rx.Observable.fromCallback(func2)(1, "");
		obsStr = Rx.Observable.fromCallback(func2, {})(1, "");
		obsStr = Rx.Observable.fromCallback(func2, {}, (results: string[]) => results[0])(1, "");

		// 3 arguments
		var func3: (a: number, b: string, c: boolean, cb: (result: string) => number) => Date = () => new Date();
		obsStr = Rx.Observable.fromCallback(func3)(1, "", true);
		obsStr = Rx.Observable.fromCallback(func3, {})(1, "", true);
		obsStr = Rx.Observable.fromCallback(func3, {}, (results: string[]) => results[0])(1, "", true);

		// multiple results
		var func0m: (cb: (result1: number, result2: number, result3: number) => void) => void = () => { };
		obsNum = Rx.Observable.fromCallback(func0m, obsStr, (results: number[]) => results[0])();
		var func1m: (a: string, cb: (result1: number, result2: number, result3: number) => void) => void = () => { };
		obsNum = Rx.Observable.fromCallback(func1m, obsStr, (results: number[]) => results[0])("");
		var func2m: (a: string, b: number, cb: (result1: string, result2: string, result3: string) => void) => void = () => { };
		obsStr = Rx.Observable.fromCallback(func2m, obsStr, (results: string[]) => results[0])("", 10);
	}

warned++;
    if (!App.isRecording && warned == 4) {
      alert('Warning: Your actions are not being recorded! Press "Start Record" or the tab key to begin recording');
    }
    var target = $(e.target);
    if (!$('.slider').length) {
      slider(target.next().html(), 230, 'auto', '#application', target.parent().parent());
    }
  });

  /**
    Board operations
   */

  var $operationSources      = '.button.draw, .button.transform, .button.misc';
  var $operationSource       = Rx.Observable.fromEventPattern(
    function addHandler(h) { $('#application').on('click', $operationSources, h) },
    function delHandler(h) { $('#application').off('click', $operationSources, h) }
  );

  var $operationSubscription = $operationSource.subscribe(function(e) {
    var target    = $(e.target).parent().attr('class').split('-');
    var targetOperation = target[0],
        targetCommand   = target[1];
    // validation error
    if ($(e.target).parent().find('[data-error]').length) {
      alert($(e.target).parent().find('[data-error]').first().attr('data-error'));
      return;
    }
    // the request
    var $command  = {
      'targetOperation': targetOperation,

var DHTSensor = require('node-dht-sensor');
var Rx = require('rx-lite');

// 大于，condi < value
Rx.Observable.prototype.max = function (condi) {
    var input = this;
    return Rx.Observable.create(function (observe) {
        var handle = input.subscribe(
            function (value) {
                if (value > condi) {
                    observe.onNext(value);
                }
            },
            function (error) {
                observe.onError(error);
            },
            function(value) {
                observe.onCompleted(value);
            }
        );
        return function () {

constructor() {
    const self = this;

    this._inputSubscription = null;
    this.setInput(new Subject());
    this._state = new Subject();
    this._output = new Subject();
    this._activators = [];
    this._paths = [];
    this._reducers = {};
    this._sagas = {};
    this._sagaMiddleware = createSagaMiddleware();
    this._store = createStore(
      (state) => state,
      applyMiddleware(this._sagaMiddleware)
    );

    // Set up internal sagas.

    // Send.
    this._sendSaga = function* sendSaga({ data, sessionId }) {
      yield self._send(data, sessionId);

module.exports = function(App) {
  var $sources = $('.function'),
      $source  = Rx.Observable.fromEvent($sources, "keypress");
  // Filter when the application is 'off'
  $source      = $source.filter(function() {
    return !$('#application').hasClass('off');
  });
  var $functionSubscription = $source.subscribe(function(e) {
    if (e.keyCode == 13) {
      var func = new Parser(e.target.value);
      try {
        func.run(); // generate parse tree
      } catch(e) {
        // syntax error
        alert("Syntax: " + e.message);
        return false;
      }
      var targetOperation = 'func',
          targetCommand   = func.identifier || 'plot';

module.exports = function(App, board) {
  var $zoomSources      = $('.zoom.in, .zoom.out');
  var $zoomSource       = Rx.Observable.fromEvent($zoomSources, "click");

  // Filter when the application is 'off'
  $zoomSource = $zoomSource.filter(function() {
    return !$('#application').hasClass('off');
  });

  var $zoomSubscription = $zoomSource.subscribe(function(e) {
    var target          = $(e.target),
        targetCommand = target.hasClass('in') ? 'zoomIn' : 'zoomOut';
    if ((targetCommand == 'zoomIn'  && board.zoomX < 5.9) ||
        (targetCommand == 'zoomOut' && board.zoomX > 0.167)) {
      var $command  = {
        'targetOperation': 'zoom',
        'targetCommand':   targetCommand,
        'command':         command['zoom'][targetCommand]
      };

module.exports = function(App) {
  /**
    Pre-queries
   */
  var $querySources  = $([
    '.circle',   '  .angle',   '.arc',
    '.ellipse',    '.segment', '.line',
     '.polygon',   '.point',   '.text',
     '.rotate',    '.reflect', '.shear',
     '.translate', '.scale',   '.delete_'
  ].join(','));
  // The query observer prepares the way for the following operations subscription
  var $querySource       = Rx.Observable.fromEvent($querySources, 'click');
  // Filter queries when the application is 'off'
  $querySource           = $querySource.filter(function() {
    return !$('#application').hasClass('off');
  });
  var warned = 0;
  var $querySubscription = $querySource.subscribe(function(e) {
    // before we do anything, warn if not recording after 4 queries
    warned++;
    if (!App.isRecording && warned == 4) {
      alert('Warning: Your actions are not being recorded! Press "Start Record" or the tab key to begin recording');
    }
    var target = $(e.target);
    if (!$('.slider').length) {
      slider(target.next().html(), 230, 'auto', '#application', target.parent().parent());
    }
  });

generatePromiseOrObservable = (json: Object): LivePromise => {
        const reqId = json.req_id.toString();

        if (this.useRx) {
            const obs = Observable.create(observer => {
                // if call is an subscription, store it in uncompleteStreamObs
                // ticks is a special case that's a stream without subscribe keyword
                if (json.subscribe || json.ticks) {
                    this.uncompleteStreamObs[reqId] = observer;
                } else {
                    this.uncompleteOneTimeObs[reqId] = observer;
                }

                return () => {
                    delete this.uncompleteOneTimeObs[reqId];
                    delete this.uncompleteStreamObs[reqId];
                };
            });

            const published = obs.publish();

var command    = require('../events/run'),
Rx             = require('rx-lite').Rx;

module.exports = function(App, board) {
  var $zoomSources      = $('.zoom.in, .zoom.out');
  var $zoomSource       = Rx.Observable.fromEvent($zoomSources, "click");

  // Filter when the application is 'off'
  $zoomSource = $zoomSource.filter(function() {
    return !$('#application').hasClass('off');
  });

  var $zoomSubscription = $zoomSource.subscribe(function(e) {
    var target          = $(e.target),
        targetCommand = target.hasClass('in') ? 'zoomIn' : 'zoomOut';
    if ((targetCommand == 'zoomIn'  && board.zoomX < 5.9) ||
        (targetCommand == 'zoomOut' && board.zoomX > 0.167)) {
      var $command  = {

var command    = require('../events/run'),
    Parser     = require('../board/functions/parser'),
    Rx         = require('rx-lite').Rx;

module.exports = function(App) {
  var $sources = $('.function'),
      $source  = Rx.Observable.fromEvent($sources, "keypress");
  // Filter when the application is 'off'
  $source      = $source.filter(function() {
    return !$('#application').hasClass('off');
  });
  var $functionSubscription = $source.subscribe(function(e) {
    if (e.keyCode == 13) {
      var func = new Parser(e.target.value);
      try {
        func.run(); // generate parse tree
      } catch(e) {
        // syntax error
        alert("Syntax: " + e.message);