@@ -2,8 +2,6 @@ import {Observable} from '../Observable';
2
2
import { Operator } from '../Operator' ;
3
3
import { Subscriber } from '../Subscriber' ;
4
4
import { Subscription } from '../Subscription' ;
5
- import { tryCatch } from '../util/tryCatch' ;
6
- import { errorObject } from '../util/errorObject' ;
7
5
import { subscribeToResult } from '../util/subscribeToResult' ;
8
6
import { OuterSubscriber } from '../OuterSubscriber' ;
9
7
@@ -39,7 +37,7 @@ export class MergeMapOperator<T, R, R2> implements Operator<T, R> {
39
37
40
38
export class MergeMapSubscriber < T , R , R2 > extends OuterSubscriber < T , R > {
41
39
private hasCompleted : boolean = false ;
42
- private buffer : T [ ] = [ ] ;
40
+ private buffer : Observable < any > [ ] = [ ] ;
43
41
private active : number = 0 ;
44
42
protected index : number = 0 ;
45
43
@@ -50,23 +48,28 @@ export class MergeMapSubscriber<T, R, R2> extends OuterSubscriber<T, R> {
50
48
super ( destination ) ;
51
49
}
52
50
53
- protected _next ( value : T ) : void {
51
+ protected _next ( value : any ) : void {
54
52
if ( this . active < this . concurrent ) {
55
- const index = this . index ++ ;
56
- const ish = tryCatch ( this . project ) ( value , index ) ;
57
- const destination = this . destination ;
58
- if ( ish === errorObject ) {
59
- destination . error ( errorObject . e ) ;
60
- } else {
61
- this . active ++ ;
62
- this . _innerSub ( ish , value , index ) ;
63
- }
53
+ this . _tryNext ( value ) ;
64
54
} else {
65
55
this . buffer . push ( value ) ;
66
56
}
67
57
}
68
58
69
- private _innerSub ( ish : Observable < R > , value : T , index : number ) : void {
59
+ protected _tryNext ( value : any ) {
60
+ let result : any ;
61
+ const index = this . index ++ ;
62
+ try {
63
+ result = this . project ( value , index ) ;
64
+ } catch ( err ) {
65
+ this . destination . error ( err ) ;
66
+ return ;
67
+ }
68
+ this . active ++ ;
69
+ this . _innerSub ( result , value , index ) ;
70
+ }
71
+
72
+ private _innerSub ( ish : any , value : T , index : number ) : void {
70
73
this . add ( subscribeToResult < T , R > ( this , ish , value , index ) ) ;
71
74
}
72
75
@@ -78,17 +81,22 @@ export class MergeMapSubscriber<T, R, R2> extends OuterSubscriber<T, R> {
78
81
}
79
82
80
83
notifyNext ( outerValue : T , innerValue : R , outerIndex : number , innerIndex : number ) : void {
81
- const { destination, resultSelector } = this ;
82
- if ( resultSelector ) {
83
- const result = tryCatch ( resultSelector ) ( outerValue , innerValue , outerIndex , innerIndex ) ;
84
- if ( result === errorObject ) {
85
- destination . error ( errorObject . e ) ;
86
- } else {
87
- destination . next ( result ) ;
88
- }
84
+ if ( this . resultSelector ) {
85
+ this . _notifyResultSelector ( outerValue , innerValue , outerIndex , innerIndex ) ;
89
86
} else {
90
- destination . next ( innerValue ) ;
87
+ this . destination . next ( innerValue ) ;
88
+ }
89
+ }
90
+
91
+ _notifyResultSelector ( outerValue : T , innerValue : R , outerIndex : number , innerIndex : number ) {
92
+ let result : any ;
93
+ try {
94
+ result = this . resultSelector ( outerValue , innerValue , outerIndex , innerIndex ) ;
95
+ } catch ( err ) {
96
+ this . destination . error ( err ) ;
97
+ return ;
91
98
}
99
+ this . destination . next ( result ) ;
92
100
}
93
101
94
102
notifyComplete ( innerSub : Subscription ) : void {
0 commit comments