alvipeo/config.js

## config.js
System.config({
  //use typescript for compilation
  transpiler: 'typescript',
  //typescript compiler options
  typescriptOptions: {
    emitDecoratorMetadata: true
  },
  //map tells the System loader where to look for things
  map: {
    app: './src'
  },
  //packages defines our app package
  packages: {
    app: {
      main: './main.ts',
      filelist: './filelest.ts',
      fileitem: './fileitem.ts',
      processingService: './processingService',
      typingText: './typingText.ts',
      defaultExtension: 'ts'
    }
  }
});

## index.html
<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <title>playground</title>
  <link rel="stylesheet" href="styles.css">
    <script src="https://code.angularjs.org/2.0.0-beta.12/angular2-polyfills.js"></script>
    <script src="https://code.angularjs.org/tools/system.js"></script>
    <script src="https://code.angularjs.org/tools/typescript.js"></script>
    <script src="config.js"></script>
    <script src="https://code.angularjs.org/2.0.0-beta.12/Rx.js"></script>
    <script src="https://code.angularjs.org/2.0.0-beta.12/angular2.dev.js"></script>
    <script src="https://code.angularjs.org/2.0.0-beta.12/http.dev.js"></script>
    <script src="https://code.angularjs.org/2.0.0-beta.12/router.dev.js"></script>
    <script>
    System.import('app')
      .catch(console.error.bind(console));
  </script>
  </head>

  <body>
    <app></app>
  </body>

</html>

## src\dummy.json
{"response": "foo"}

## src\fileitem.ts
export class IlgFileItem {
  private _progress: number = 0;

  constructor (private _id: number) {}

  get progress(): number {
    return this._progress;
  }

  updateProgress(): boolean {
    if (this._progress < 100) {
      this._progress += 5;
      return true;
    }
    return false;
  }
}

## src\filelist.ts
import {Input, Component, ChangeDetectionStrategy, ChangeDetectorRef, AfterViewInit, OnChanges, SimpleChange, ViewChild} from "angular2/core";
import {Observable} from 'rxjs/Rx';
import {IlgFlyingText} from './typingText';
import {IlgFileItem} from './fileitem';

@Component({
  selector: "ilg-filelist",
  template: `<div>
    <ul>
    <li class="progr" *ngFor="#file of items" [style.width.%]="file.progress"><span>{{file.progress}}</span></li>
    </ul>
            </div>`,
  //changeDetection: ChangeDetectionStrategy.OnPush // COMMENT OUT THIS LINE TO MAKE EVERYTHING WORK
})
export class IlgFilelist {
  @Input() items: Array<IlgFileItem>;

  constructor() {
  }
}

## src\main.ts
import {Component} from 'angular2/core';
import {bootstrap} from 'angular2/platform/browser';
import {Observable} from 'rxjs/Rx';
import {IlgFilelist} from "./filelist";
import {ProcessingService} from './processingService';

@Component({
    selector: 'app',
    template: `<div>
      <h2>Fixed.</h2>
      <hr/>
      <p>Please open a console to see the result. The question is regarding RxJS operators of processingService (see processingService.ts).</p>
      <p><strong>Why it doesn't get to mergeMap on every array item, but first the whole array is processed in map() operator (1 by 1) and only after that it goes to mergeMap() ?</strong></p>

      <a (click)="$event.preventDefault(); run();" href="#">Start process</a>
      <hr/>
      <ilg-filelist [items]="_items"></ilg-filelist>
      <hr/>
    </div>`,
    directives: [IlgFilelist]
})
export class App {
  constructor() {}

  run() {
    this._processingService = new ProcessingService(5);

    this._processingService.process()
      .subscribe((array: Array<IlgFileItem) => {
        if (this._items === array) {
          console.error("Arrays are equal");
        }
        this._items = array;
        //console.log("array", array);
        //console.log(Zone.current.name);
      },
      null,
      () => console.log("Done: ", Zone.current.name);
  }
}

bootstrap(App, []);

## src\processingService.ts
import {Observable} from 'rxjs/Rx';
import {IlgFileItem} from './fileitem';

export class ProcessingService {
  constructor(private _amount: number) {

  }

  process(): Observable<Array<IlgFileItem>> {
    const arr = Array<IlgFileItem>();
    for (let i=0; i<this._amount; i++) {
      arr[i] = new IlgFileItem(i));
    }

    const dummyArr = Array.from({length: 100}, (v, k) => k);

    const ob$ =
    Observable.from(dummyArr)
    //.do(this.simulateProcessing)
    .map(x => arr)
    .mergeMap(arr => {
        console.log("::: step 1", arr);

        this.simulateProcessing();

        return Observable.fromPromise(fetch("src/dummy.json")).zip(Observable.of(arr));
      }, null, 1)
      .mergeMap(value => {
        console.log(":: step 2");

        return Observable.of(value[1]);
      }, (outer, inner) => { return inner; })
      .map(arr => {
        do {
          let rndIdx = this.getRandomIndex();
        } while (!arr[rndIdx].updateProgress());

        return new Array<IlgFileItem>(...arr);
      })
      .takeWhile(arr => !arr.every((item: IlgFileItem) => item.progress == 100));

      /*
  const ob$ =
    Observable.from(dummyArr)
    .concatMap((x) => Observable.of(x)
      //.do(this.simulateProcessing)
      .map(x => arr)
      .map(arr => {
          console.log("::: step 1");
          do {
            let rndIdx = this.getRandomIndex();
            //console.log("update progress for " + rndIdx);
          } while (!arr[rndIdx].updateProgress());

          const newArr = new Array<IlgFileItem>(...arr);
          newArr.forEach((element) => console.log(element));
          console.log("-----------------");

          return [fetch("src/dummy.json"), newArr];
        })
        .mergeMap(value => { console.log(":: step 2"); return value[0]; },
        (arr, inner) => { console.log("inside", arr); return arr; }, 1)
        .map(val => { console.log(":: step 3", val[1]); return val[1]; })
        .takeWhile(arr => !arr.every((item: IlgFileItem) => item.progress == 100))
      );
      */

    return ob$;
  }

  private simulateProcessing() {
    for (let t = 0; t < 1000; t++) {
    let i = 0;
      (function(Pi,n){
        Pi += (4/n) - (4/(n+2));
        n += 4;
        i++;
        if (i <= 3000) { // 27000
          arguments.callee(Pi, n);
        }
      })(0, 1);
    }
  }

  private getRandomIndex(): number {
    const min = 0;
    const max = this._amount;
    return Math.floor(Math.random() * (max - min)) + min;
  }
}

## src\typingText.ts
import {Input, Component, ChangeDetectionStrategy, ChangeDetectorRef, AfterViewInit, OnChanges, SimpleChange} from "angular2/core";
import {Observable} from 'rxjs/Rx';

@Component({
    selector: "ilg-flying-text",
    template: "{{ showError$ | async }}",
    changeDetection: ChangeDetectionStrategy.OnPush
})
export class IlgFlyingText implements AfterViewInit, OnChanges {
    @Input() text: string;

    showError$: Observable<string>;

    ngAfterViewInit() {
        this.initObservable();
    }

    ngOnChanges(changes: { [propName: string]: SimpleChange }) {
        this.initObservable();
    }

    private initObservable() {
        if (!!this.text) {
            console.log(this.text);
            this.showError$ = this.createErrorObservable(this.text);
        } else {
            this.showError$ = Observable.of(null);
        }
    }

    private createErrorObservable(errText: string): Observable<string> {
        //console.log(err);
        let o1$ = Observable.of(errText);
        let o2$ = Observable.interval(25);
        let o$ = o1$
            .switch()
            .zip(o2$)
            .map(tuple => tuple[0])
            .scan((acc, val) => acc += val[0]);
        //.do(val => console.log(val));
        return o$;
    }
}

## styles.css
/* todo: add styles */
.progr {
  height: 2rem;
  background-color: blue;
  color: white;
  text-align: center;
  display: table;
}

.progr span {
  display: table-cell;
  vertical-align: middle;
  padding: 0.3rem;
}
	System.config({
	//use typescript for compilation
	transpiler: 'typescript',
	//typescript compiler options
	typescriptOptions: {
	emitDecoratorMetadata: true
	},
	//map tells the System loader where to look for things
	map: {
	app: './src'
	},
	//packages defines our app package
	packages: {
	app: {
	main: './main.ts',
	filelist: './filelest.ts',
	fileitem: './fileitem.ts',
	processingService: './processingService',
	typingText: './typingText.ts',
	defaultExtension: 'ts'
	}
	}
	});
	<!doctype html>
	<html lang="en">
	<head>
	<meta charset="utf-8">
	<title>playground</title>
	<link rel="stylesheet" href="styles.css">
	<script src="https://code.angularjs.org/2.0.0-beta.12/angular2-polyfills.js"></script>
	<script src="https://code.angularjs.org/tools/system.js"></script>
	<script src="https://code.angularjs.org/tools/typescript.js"></script>
	<script src="config.js"></script>
	<script src="https://code.angularjs.org/2.0.0-beta.12/Rx.js"></script>
	<script src="https://code.angularjs.org/2.0.0-beta.12/angular2.dev.js"></script>
	<script src="https://code.angularjs.org/2.0.0-beta.12/http.dev.js"></script>
	<script src="https://code.angularjs.org/2.0.0-beta.12/router.dev.js"></script>
	<script>
	System.import('app')
	.catch(console.error.bind(console));
	</script>
	</head>

	<body>
	<app></app>
	</body>

	</html>
	export class IlgFileItem {
	private _progress: number = 0;

	constructor (private _id: number) {}

	get progress(): number {
	return this._progress;
	}

	updateProgress(): boolean {
	if (this._progress < 100) {
	this._progress += 5;
	return true;
	}
	return false;
	}
	}
	import {Input, Component, ChangeDetectionStrategy, ChangeDetectorRef, AfterViewInit, OnChanges, SimpleChange, ViewChild} from "angular2/core";
	import {Observable} from 'rxjs/Rx';
	import {IlgFlyingText} from './typingText';
	import {IlgFileItem} from './fileitem';

	@Component({
	selector: "ilg-filelist",
	template: `<div>
	<ul>
	<li class="progr" *ngFor="#file of items" [style.width.%]="file.progress"><span>{{file.progress}}</span></li>
	</ul>
	</div>`,
	//changeDetection: ChangeDetectionStrategy.OnPush // COMMENT OUT THIS LINE TO MAKE EVERYTHING WORK
	})
	export class IlgFilelist {
	@Input() items: Array<IlgFileItem>;

	constructor() {
	}
	}
	import {Component} from 'angular2/core';
	import {bootstrap} from 'angular2/platform/browser';
	import {Observable} from 'rxjs/Rx';
	import {IlgFilelist} from "./filelist";
	import {ProcessingService} from './processingService';

	@Component({
	selector: 'app',
	template: `<div>
	<h2>Fixed.</h2>
	<hr/>
	<p>Please open a console to see the result. The question is regarding RxJS operators of processingService (see processingService.ts).</p>
	<p><strong>Why it doesn't get to mergeMap on every array item, but first the whole array is processed in map() operator (1 by 1) and only after that it goes to mergeMap() ?</strong></p>

	<a (click)="$event.preventDefault(); run();" href="#">Start process</a>
	<hr/>
	<ilg-filelist [items]="_items"></ilg-filelist>
	<hr/>
	</div>`,
	directives: [IlgFilelist]
	})
	export class App {
	constructor() {}

	run() {
	this._processingService = new ProcessingService(5);

	this._processingService.process()
	.subscribe((array: Array<IlgFileItem) => {
	if (this._items === array) {
	console.error("Arrays are equal");
	}
	this._items = array;
	//console.log("array", array);
	//console.log(Zone.current.name);
	},
	null,
	() => console.log("Done: ", Zone.current.name);
	}
	}

	bootstrap(App, []);
	import {Observable} from 'rxjs/Rx';
	import {IlgFileItem} from './fileitem';

	export class ProcessingService {
	constructor(private _amount: number) {

	}

	process(): Observable<Array<IlgFileItem>> {
	const arr = Array<IlgFileItem>();
	for (let i=0; i<this._amount; i++) {
	arr[i] = new IlgFileItem(i));
	}

	const dummyArr = Array.from({length: 100}, (v, k) => k);

	const ob$ =
	Observable.from(dummyArr)
	//.do(this.simulateProcessing)
	.map(x => arr)
	.mergeMap(arr => {
	console.log("::: step 1", arr);

	this.simulateProcessing();

	return Observable.fromPromise(fetch("src/dummy.json")).zip(Observable.of(arr));
	}, null, 1)
	.mergeMap(value => {
	console.log(":: step 2");

	return Observable.of(value[1]);
	}, (outer, inner) => { return inner; })
	.map(arr => {
	do {
	let rndIdx = this.getRandomIndex();
	} while (!arr[rndIdx].updateProgress());

	return new Array<IlgFileItem>(...arr);
	})
	.takeWhile(arr => !arr.every((item: IlgFileItem) => item.progress == 100));

	/*
	const ob$ =
	Observable.from(dummyArr)
	.concatMap((x) => Observable.of(x)
	//.do(this.simulateProcessing)
	.map(x => arr)
	.map(arr => {
	console.log("::: step 1");
	do {
	let rndIdx = this.getRandomIndex();
	//console.log("update progress for " + rndIdx);
	} while (!arr[rndIdx].updateProgress());

	const newArr = new Array<IlgFileItem>(...arr);
	newArr.forEach((element) => console.log(element));
	console.log("-----------------");

	return [fetch("src/dummy.json"), newArr];
	})
	.mergeMap(value => { console.log(":: step 2"); return value[0]; },
	(arr, inner) => { console.log("inside", arr); return arr; }, 1)
	.map(val => { console.log(":: step 3", val[1]); return val[1]; })
	.takeWhile(arr => !arr.every((item: IlgFileItem) => item.progress == 100))
	);
	*/

	return ob$;
	}

	private simulateProcessing() {
	for (let t = 0; t < 1000; t++) {
	let i = 0;
	(function(Pi,n){
	Pi += (4/n) - (4/(n+2));
	n += 4;
	i++;
	if (i <= 3000) { // 27000
	arguments.callee(Pi, n);
	}
	})(0, 1);
	}
	}

	private getRandomIndex(): number {
	const min = 0;
	const max = this._amount;
	return Math.floor(Math.random() * (max - min)) + min;
	}
	}
	/* todo: add styles */
	.progr {
	height: 2rem;
	background-color: blue;
	color: white;
	text-align: center;
	display: table;
	}

	.progr span {
	display: table-cell;
	vertical-align: middle;
	padding: 0.3rem;
	}