import { Operator } from '../Operator';
import { Subscriber } from '../Subscriber';
import { Subscription } from '../Subscription';
import { Operator } from '../Operator';
import { Subscriber } from '../Subscriber';
import { Subscription } from '../Subscription';
import { MonoTypeOperatorFunction, OperatorFunction, ObservableInput, SchedulerLike } from '../types';
import { MonoTypeOperatorFunction, OperatorFunction, ObservableInput, SchedulerLike } from '../types';
/* tslint:disable:max-line-length */
export function expand<T, R>(project: (value: T, index: number) => ObservableInput<R>, concurrent?: number, scheduler?: SchedulerLike): OperatorFunction<T, R>;
/* tslint:disable:max-line-length */
export function expand<T, R>(project: (value: T, index: number) => ObservableInput<R>, concurrent?: number, scheduler?: SchedulerLike): OperatorFunction<T, R>;
*/
export function expand<T, R>(project: (value: T, index: number) => ObservableInput<R>,
concurrent: number = Number.POSITIVE_INFINITY,
*/
export function expand<T, R>(project: (value: T, index: number) => ObservableInput<R>,
concurrent: number = Number.POSITIVE_INFINITY,
concurrent = (concurrent || 0) < 1 ? Number.POSITIVE_INFINITY : concurrent;
return (source: Observable<T>) => source.lift(new ExpandOperator(project, concurrent, scheduler));
concurrent = (concurrent || 0) < 1 ? Number.POSITIVE_INFINITY : concurrent;
return (source: Observable<T>) => source.lift(new ExpandOperator(project, concurrent, scheduler));
export class ExpandOperator<T, R> implements Operator<T, R> {
constructor(private project: (value: T, index: number) => ObservableInput<R>,
private concurrent: number,
export class ExpandOperator<T, R> implements Operator<T, R> {
constructor(private project: (value: T, index: number) => ObservableInput<R>,
private concurrent: number,
private index: number = 0;
private active: number = 0;
private hasCompleted: boolean = false;
private index: number = 0;
private active: number = 0;
private hasCompleted: boolean = false;
constructor(destination: Subscriber<R>,
private project: (value: T, index: number) => ObservableInput<R>,
private concurrent: number,
constructor(destination: Subscriber<R>,
private project: (value: T, index: number) => ObservableInput<R>,
private concurrent: number,
try {
const { project } = this;
const result = project(value, index);
try {
const { project } = this;
const result = project(value, index);
} else {
const state: DispatchArg<T, R> = { subscriber: this, result, value, index };
const destination = this.destination as Subscription;
} else {
const state: DispatchArg<T, R> = { subscriber: this, result, value, index };
const destination = this.destination as Subscription;
- destination.add(this.scheduler.schedule<DispatchArg<T, R>>(ExpandSubscriber.dispatch, 0, state));
+ destination.add(this.scheduler.schedule<DispatchArg<T, R>>(ExpandSubscriber.dispatch as any, 0, state));
}
}
private subscribeToProjection(result: any, value: T, index: number): void {
this.active++;
const destination = this.destination as Subscription;
}
}
private subscribeToProjection(result: any, value: T, index: number): void {
this.active++;
const destination = this.destination as Subscription;
- destination.add(subscribeToResult<T, R>(this, result, value, index));
+ destination.add(innerSubscribe(result, new SimpleInnerSubscriber(this)));
}
protected _complete(): void {
this.hasCompleted = true;
if (this.hasCompleted && this.active === 0) {
}
protected _complete(): void {
this.hasCompleted = true;
if (this.hasCompleted && this.active === 0) {
- notifyNext(outerValue: T, innerValue: R,
- outerIndex: number, innerIndex: number,
- innerSub: InnerSubscriber<T, R>): void {
+ notifyNext(innerValue: R): void {
this.active--;
if (buffer && buffer.length > 0) {
this._next(buffer.shift());
}
if (this.hasCompleted && this.active === 0) {
this.active--;
if (buffer && buffer.length > 0) {
this._next(buffer.shift());
}
if (this.hasCompleted && this.active === 0) {