angular.module('app', [])

.controller('MyCtrl', function($scope, $http){

  $scope.timer = Rx.Observable

    .interval(1000)

    .take(50);

  $scope.repos = $http.get('https://api.github.com/orgs/Reactive-Extensions/repos');

​

})

.filter('async', function() {

  const values = {};

  const subscriptions = {};

​

  function async(input, scope) {

    // Make sure we have an Observable or a Promise

    if (!input || !(input.subscribe || input.then)) {

      return;

    }

​

    const inputId = objectId(input);

    if (!(inputId in subscriptions)) {

      const subscriptionStrategy = input.subscribe && input.subscribe.bind(input)

      || input.success && input.success.bind(input) // So it works with HttpPromise

      || input.then.bind(input);

​

      subscriptions[inputId] = subscriptionStrategy(value => {

        values[inputId] = value;

        if (scope && scope.$applyAsync) {

          scope.$applyAsync();

        }

      });

    }

​

    return values[inputId] || undefined;

  };

​

  // Need a way to tell the input objects apart from each other (so we only subscribe to them once)

  let nextObjectID = 0;

  function objectId(obj) {

    if (!obj.hasOwnProperty('__asyncFilterObjectID__')) {

      obj.__asyncFilterObjectID__ = ++nextObjectID;

    }

​

    return obj.__asyncFilterObjectID__;

  }

​

  // So that Angular does not cache the return value

  async.$stateful = true;

​

  return async;

});