By returning an Observable, we can either hand back a static array or an infinite stream over time (every 60 seconds)
429461
function getDemoActions() {
    if (infinite) {
      const faker = require("faker")
      const controls = [
        map(() => ({
          payload: {
            toSpeak: faker.company.catchPhrase()
          }
        }))
      ]
      if (count) {
        controls.push(take(count))
      }
      return interval(1000).pipe(...controls)
    }

    return interval(250).pipe(
      zip(sayings, (_, toSpeak) => ({
        payload: {
          toSpeak
        }
      })),
      take(count)
    )
  }
697759
function SyncFuture(timeout)
{
    var self = this;
    
    this.resolved = false;
    this.fiber = Fiber.current;
    this.yielding = false;
    this.timeout = timeout;
    this.time = null;
    
    this._timeoutId = null;
    this._result = undefined;
    this._error = null;
    this._start = +new Date;
    
    Sync.stat.totalFutures++;
    Sync.stat.activeFutures++;
    
    // Create timeout error to capture stack trace correctly
    self.timeoutError = new Error();
    Error.captureStackTrace(self.timeoutError, arguments.callee);
    
    this.ticket = function Future()
    {
        // clear timeout if present
        if (self._timeoutId) clearTimeout(self._timeoutId);
        // measure time
        self.time = new Date - self._start;
        
        // forbid to call twice
        if (self.resolved) return;
        self.resolved = true;

        // err returned as first argument
        var err = arguments[0];
        if (err) {
            self._error = err;
        }
        else {
            self._result = arguments[1];
        }
        
        // remove self from current fiber
        self.fiber.removeFuture(self.ticket);
        Sync.stat.activeFutures--;
        
        if (self.yielding && Fiber.current !== self.fiber) {
            self.yielding = false;
            self.fiber.run();
        }
        else if (self._error) {
            throw self._error;
        }
    }
    
    this.ticket.__proto__ = this;
    
    this.ticket.yield = function() {
        while (!self.resolved) {
            self.yielding = true;
            if (self.timeout) {
                self._timeoutId = setTimeout(function(){
                    self.timeoutError.message = 'Future function timed out at ' + self.timeout + ' ms';
                    self.ticket(self.timeoutError);
                }, self.timeout)
            }
            Fiber.yield();
        }
        if (self._error) throw self._error;
        return self._result;
    }
    
    this.ticket.__defineGetter__('result', function(){
        return this.yield();
    });
    
    this.ticket.__defineGetter__('error', function(){
        if (self._error) {
            return self._error;
        }
        try {
            this.result;
        }
        catch (e) {
            return e;
        }
        return null;
    });
    
    this.ticket.__defineGetter__('timeout', function(){
        return self.timeout;
    });
    
    this.ticket.__defineSetter__('timeout', function(value){
        self.timeout = value;
    });
    
    // append self to current fiber
    this.fiber.addFuture(this.ticket);
    
    return this.ticket;
}
1170058
function () {
            return new me.ObservableVector2d(Math.floor(this._x), Math.floor(this._y), {onUpdate: this.onUpdate});
        }
1002673
public static Observable<byte[]> from(InputStream is, int size) {
        return Observable.create(new OnSubscribeInputStream(is, size));
    }
166793
function (obj, resolve, reject) {
        // if promise like
        if (obj && typeof obj.then === 'function') {
            return Promise.resolve(obj).then(resolve).catch(reject);
        }
        // if observable
        if (obj && typeof obj.subscribe === 'function') {
            var obs_1 = obj;
            return new SafeSubscription(function (sub) { return obs_1.subscribe({
                next: function () {
                    if (sub.subscribed) {
                        sub.unsubscribe();
                        resolve(obj);
                    }
                },
                error: function (error) {
                    if (sub.subscribed) {
                        sub.unsubscribe();
                        reject(error);
                    }
                },
                complete: function () {
                    if (sub.subscribed) {
                        sub.unsubscribe();
                        resolve(obj);
                    }
                }
            }); });
        }
        // else just resolve
        try {
            return resolve(obj);
        }
        catch (error) {
            return reject(error);
        }
    }
1242646
public static <T> ServiceFuture<T> fromResponse(final Observable<ServiceResponse<T>> observable) {
        final ServiceFuture<T> serviceFuture = new ServiceFuture<>();
        serviceFuture.subscription = observable
            .last()
            .subscribe(new Action1<ServiceResponse<T>>() {
                @Override
                public void call(ServiceResponse<T> t) {
                    serviceFuture.set(t.body());
                }
            }, new Action1<Throwable>() {
                @Override
                public void call(Throwable throwable) {
                    serviceFuture.setException(throwable);
                }
            });
        return serviceFuture;
    }
1334912
private static Observable<Map<String, Object>> tombstone(Observable<Map<String, Object>> instanceStream, InstanceKey instanceKey) {
        return instanceStream.publish(is -> {
            Observable<Map<String, Object>> tombstone = is
                    // collect all unique "TypeAndName" keys
                    .collect(() -> new HashSet<TypeAndNameKey>(), (listOfTypeAndName, event) -> {
                        listOfTypeAndName.add((TypeAndNameKey) event.get("TypeAndName"));
                    })
                    // when instanceStream completes, emit a "tombstone" for each "TypeAndName" in the HashSet collected above
                    .flatMap(listOfTypeAndName -> {
                        return Observable.from(listOfTypeAndName)
                                .map(typeAndName -> {
                                    Map<String, Object> tombstoneForTypeAndName = new LinkedHashMap<>();
                                    tombstoneForTypeAndName.put("tombstone", "true");
                                    tombstoneForTypeAndName.put("InstanceKey", instanceKey);
                                    tombstoneForTypeAndName.put("TypeAndName", typeAndName);
                                    tombstoneForTypeAndName.put("name", typeAndName.getName());
                                    tombstoneForTypeAndName.put("type", typeAndName.getType());
                                    return tombstoneForTypeAndName;
                                });
                    });
            
            // concat the tombstone events to the end of the original stream
            return is.mergeWith(tombstone);
        });
    }
744219
function timeoutWith$2(due, withObservable, scheduler) {
    if (scheduler === void 0) { scheduler = async.async; }
    var absoluteTimeout = isDate_1.isDate(due);
    var waitFor = absoluteTimeout ? (+due - scheduler.now()) : Math.abs(due);
    return this.lift(new TimeoutWithOperator(waitFor, absoluteTimeout, withObservable, scheduler));
}
879373
<E> Observable<E> wrapObservable(Observable<E> obs) {
        return obs.subscribeOn(Schedulers.io()).observeOn(Schedulers.io());
    }
505053
function fromIterable(input, scheduler) {
    if (!input) {
        throw new Error('Iterable cannot be null');
    }
    if (!scheduler) {
        return new _Observable__WEBPACK_IMPORTED_MODULE_0__["Observable"](Object(_util_subscribeToIterable__WEBPACK_IMPORTED_MODULE_3__["subscribeToIterable"])(input));
    }
    else {
        return new _Observable__WEBPACK_IMPORTED_MODULE_0__["Observable"](function (subscriber) {
            var sub = new _Subscription__WEBPACK_IMPORTED_MODULE_1__["Subscription"]();
            var iterator;
            sub.add(function () {
                if (iterator && typeof iterator.return === 'function') {
                    iterator.return();
                }
            });
            sub.add(scheduler.schedule(function () {
                iterator = input[_symbol_iterator__WEBPACK_IMPORTED_MODULE_2__["iterator"]]();
                sub.add(scheduler.schedule(function () {
                    if (subscriber.closed) {
                        return;
                    }
                    var value;
                    var done;
                    try {
                        var result = iterator.next();
                        value = result.value;
                        done = result.done;
                    }
                    catch (err) {
                        subscriber.error(err);
                        return;
                    }
                    if (done) {
                        subscriber.complete();
                    }
                    else {
                        subscriber.next(value);
                        this.schedule();
                    }
                }));
            }));
            return sub;
        });
    }
}
867947
function onErrorResumeNextStatic() {
    var nextSources = [];
    for (var _i = 0; _i < arguments.length; _i++) {
        nextSources[_i - 0] = arguments[_i];
    }
    var source = null;
    if (nextSources.length === 1 && isArray.isArray(nextSources[0])) {
        nextSources = nextSources[0];
    }
    source = nextSources.shift();
    return new FromObservable_1.FromObservable(source, null).lift(new OnErrorResumeNextOperator(nextSources));
}
881304
function onErrorResumeNext() {
    var sources = [];
    for (var _i = 0; _i < arguments.length; _i++) {
        sources[_i] = arguments[_i];
    }
    if (sources.length === 0) {
        return _empty__WEBPACK_IMPORTED_MODULE_3__["EMPTY"];
    }
    var first = sources[0], remainder = sources.slice(1);
    if (sources.length === 1 && Object(_util_isArray__WEBPACK_IMPORTED_MODULE_2__["isArray"])(first)) {
        return onErrorResumeNext.apply(void 0, first);
    }
    return new _Observable__WEBPACK_IMPORTED_MODULE_0__["Observable"](function (subscriber) {
        var subNext = function () { return subscriber.add(onErrorResumeNext.apply(void 0, remainder).subscribe(subscriber)); };
        return Object(_from__WEBPACK_IMPORTED_MODULE_1__["from"])(first).subscribe({
            next: function (value) { subscriber.next(value); },
            error: subNext,
            complete: subNext,
        });
    });
}
359585
static <T> Observable<T> wrap(
            ObservableSource<T> source, Collection<ReactiveInstrumenter> instrumentations) {
        if (source instanceof Callable) {
            return new RxInstrumentedCallableObservable<>(source, instrumentations);
        }
        return new RxInstrumentedObservable<>(source, instrumentations);
    }
75676
public Response get(HttpClientRequest<ByteBuf> req, long timeout, TimeUnit timeUnit) {
    final String uri = req.getUri();
    final Response result = new Response();
    try {
      final Func1<HttpClientResponse<ByteBuf>, Observable<byte[]>> process = response -> {
        result.status = response.getStatus().code();
        result.headers = response.getHeaders();
        final Func2<ByteArrayOutputStream, ByteBuf, ByteArrayOutputStream>
            accumulator = (baos, bb) -> {
          try {
            bb.readBytes(baos, bb.readableBytes());
          } catch (IOException e) {
            throw new RuntimeException(e);
          }
          return baos;
        };
        return response.getContent()
            .reduce(new ByteArrayOutputStream(), accumulator)
            .map(ByteArrayOutputStream::toByteArray);
      };

      result.body = rxHttp.submit(req)
          .flatMap(process)
          .subscribeOn(Schedulers.io())
          .toBlocking()
          .toFuture()
          .get(timeout, timeUnit);
      return result;
    } catch (Exception e) {
      throw new RuntimeException("failed to get url: " + uri, e);
    }
  }
1190762
function Sustained (obs, timeThreshold, checkUpdateImmediately) {
  var outputValue = Value(obs())
  var lastValue = null
  var timer = null

  return computed(outputValue, v => v, {
    onListen: () => watch(obs, onChange)
  })

  function onChange (value) {
    if (checkUpdateImmediately && checkUpdateImmediately(value)) {
      clearTimeout(timer)
      update()
    } else if (value !== lastValue) {
      clearTimeout(timer)
      var delay = typeof timeThreshold === 'function' ? timeThreshold(value, outputValue()) : timeThreshold
      timer = setTimeout(update, delay)
    }
    lastValue = value
  }

  function update () {
    var value = obs()
    if (value !== outputValue()) {
      outputValue.set(value)
    }
  }
}
849139
function Stream(options) {

  /**
   * Whether or not the Stream was initialized as
   * a single- or multi-key stream.
   *
   * @property multi
   * @type Boolean
   */
  reader(this, 'multi', !(options && options.apiKey));

  if (!this.multi) {
    options.headers = options.headers || {};
    options.headers['Zotero-API-Key'] = options.apiKey;

    delete options.apiKey;
  }

  Client.call(this, options);

  /**
   * @property retry
   * @type Object
   */
  reader(this, 'retry', {
    delay: Stream.DEFAULT_RETRY
  });

  /**
   * @property idle
   * @type Object
   */
  reader(this, 'idle', {
    ping: Stream.IDLE_TIMEOUT,
    pong: Stream.IDLE_TIMEOUT / 2
  });

  /**
   * @property subscriptions
   * @type Subscriptions
   */
  this.subscriptions = new Subscriptions();

  this.open = this.open.bind(this);
  this.opened = this.opened.bind(this);
  this.closed = this.closed.bind(this);
  this.ping = this.ping.bind(this);
  this.pong = this.pong.bind(this);
  this.receive = this.receive.bind(this);
  this.error = this.error.bind(this);

  this.open();
}
540893
@SuppressWarnings("unused")
    private static Observable<GroupedObservable<TypeAndNameKey, Map<String, Object>>> aggregateUsingFlattenedGroupBy(Observable<GroupedObservable<InstanceKey, Map<String, Object>>> stream) {
        Observable<Map<String, Object>> allData = stream.flatMap(instanceStream -> {
            return instanceStream
                    .map((Map<String, Object> event) -> {
                        event.put("InstanceKey", instanceStream.getKey());
                        event.put("TypeAndName", TypeAndNameKey.from(String.valueOf(event.get("type")), String.valueOf(event.get("name"))));
                        return event;
                    })
                    .compose(is -> {
                        return tombstone(is, instanceStream.getKey());
                    });
        });

        Observable<GroupedObservable<TypeAndNameKey, Map<String, Object>>> byCommand = allData
                .groupBy((Map<String, Object> event) -> {
                    return (TypeAndNameKey) event.get("TypeAndName");
                });

        return byCommand
                .map(commandGroup -> {
                    Observable<Map<String, Object>> sumOfDeltasForAllInstancesForCommand = commandGroup
                            .groupBy((Map<String, Object> json) -> {
                                return json.get("InstanceKey");
                            }).flatMap(instanceGroup -> {
                                // calculate and output deltas for each instance stream per command
                                    return instanceGroup
                                            .takeUntil(d -> d.containsKey("tombstone"))
                                            .startWith(Collections.<String, Object> emptyMap())
                                            .map(data -> {
                                                if (data.containsKey("tombstone")) {
                                                    return Collections.<String, Object> emptyMap();
                                                } else {
                                                    return data;
                                                }
                                            })
                                            .buffer(2, 1)
                                            .filter(list -> list.size() == 2)
                                            .map(StreamAggregator::previousAndCurrentToDelta)
                                            .filter(data -> data != null && !data.isEmpty());
                                })
                            // we now have all instance deltas merged into a single stream per command
                            // and sum them into a single stream of aggregate values
                            .scan(new HashMap<String, Object>(), StreamAggregator::sumOfDelta)
                            .skip(1);

                    // we artificially wrap in a GroupedObservable that communicates the CommandKey this stream represents
                return GroupedObservable.from(commandGroup.getKey(), sumOfDeltasForAllInstancesForCommand);
            });
    }