From 297ff8a8986b30fad2b91f2e46e6eacd1a40b42d Mon Sep 17 00:00:00 2001
From: Nairi Narinyan <nairi.narinyan@gmail.com>
Date: Thu, 29 Aug 2019 15:03:49 +0400
Subject: [PATCH] add ExecSubject and generic executor

---
 .gitignore        |  1 +
 package-lock.json | 14 +++++++-------
 package.json      |  6 +++---
 src/executors.ts  | 44 ++++++++++++++++++++++++++++++++++++++++++--
 4 files changed, 53 insertions(+), 12 deletions(-)

diff --git a/.gitignore b/.gitignore
index 8b035cb..dcb39ca 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,6 +1,7 @@
 node_modules
 build
 notes.md
+examples
 *.js
 *.d.ts
 *.tgz
diff --git a/package-lock.json b/package-lock.json
index d5d86c0..4661896 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -1,6 +1,6 @@
 {
   "name": "reactive-blocs",
-  "version": "0.0.6",
+  "version": "0.0.11",
   "lockfileVersion": 1,
   "requires": true,
   "dependencies": {
@@ -77,9 +77,9 @@
       "dev": true
     },
     "rxjs": {
-      "version": "6.5.1",
-      "resolved": "https://registry.npmjs.org/rxjs/-/rxjs-6.5.1.tgz",
-      "integrity": "sha512-y0j31WJc83wPu31vS1VlAFW5JGrnGC+j+TtGAa1fRQphy48+fDYiDmX8tjGloToEsMkxnouOg/1IzXGKkJnZMg==",
+      "version": "6.5.2",
+      "resolved": "https://registry.npmjs.org/rxjs/-/rxjs-6.5.2.tgz",
+      "integrity": "sha512-HUb7j3kvb7p7eCUHE3FqjoDsC1xfZQ4AHFWfTKSpZ+sAhhz5X1WX0ZuUqWbzB2QhSLp3DoLUG+hMdEDKqWo2Zg==",
       "dev": true,
       "requires": {
         "tslib": "^1.9.0"
@@ -96,9 +96,9 @@
       }
     },
     "tslib": {
-      "version": "1.9.3",
-      "resolved": "https://registry.npmjs.org/tslib/-/tslib-1.9.3.tgz",
-      "integrity": "sha512-4krF8scpejhaOgqzBEcGM7yDIEfi0/8+8zDRZhNZZ2kjmHJ4hv3zCbQWxoJGz1iw5U0Jl0nma13xzHXcncMavQ==",
+      "version": "1.10.0",
+      "resolved": "https://registry.npmjs.org/tslib/-/tslib-1.10.0.tgz",
+      "integrity": "sha512-qOebF53frne81cf0S9B41ByenJ3/IuH8yJKngAX35CmiZySA0khhkovshKK+jGCaMnVomla7gVlIcc3EvKPbTQ==",
       "dev": true
     },
     "typescript": {
diff --git a/package.json b/package.json
index a6fa622..facd6a6 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
     "name": "reactive-blocs",
-    "version": "0.0.11",
+    "version": "0.0.20",
     "description": "Reactive Blocs",
     "main": "index.js",
     "scripts": {
@@ -22,13 +22,13 @@
     },
     "homepage": "https://github.com/nairinarinyan/reactive-blocs#readme",
     "peerDependencies": {
-        "rxjs": "^6.5.1",
+        "rxjs": "^6.5.2",
         "react": "^16.8.6"
     },
     "devDependencies": {
         "@types/react": "^16.8.14",
         "react": "^16.8.6",
-        "rxjs": "^6.5.1",
+        "rxjs": "^6.5.2",
         "typescript": "^3.4.2"
     }
 }
diff --git a/src/executors.ts b/src/executors.ts
index 0f52653..d196666 100644
--- a/src/executors.ts
+++ b/src/executors.ts
@@ -1,5 +1,5 @@
-import { Observable, BehaviorSubject, of } from 'rxjs';
-import { distinctUntilChanged } from 'rxjs/operators';
+import { Observable, BehaviorSubject, of, Subject, merge, Subscriber, Subscription, SubscriptionLike, throwError } from 'rxjs';
+import { distinctUntilChanged, mergeMap, catchError, share, switchMap, filter, tap } from 'rxjs/operators';
 
 export type Executor<T> = (...execArgs: any[]) => Observable<T>;
 export type ExecPerformer = (...execArgs: any[]) => void;
@@ -43,3 +43,43 @@ export function execControlled<T>(exec: Executor<T>, control: Control, initialVa
 
     return subject;
 }
+
+export class ExecSubject<T> extends BehaviorSubject<T> {
+    private _executionStream = new Subject<any>();
+
+    constructor(
+        private _executor: Executor<T>,
+        _initialValue?: T,
+        private _error ?: BehaviorSubject<Error>,
+    ) {
+        super(_initialValue);
+        this.init();
+    }
+
+    private init() {
+        this._executionStream.pipe(
+            switchMap(args => this.performExec(this._executor, args)),
+            tap(error => {
+                if (error instanceof Error && this._error) {
+                    this._error.next(error);
+                }
+            }),
+            filter(result => !(result instanceof Error)),
+            share(),
+        ).subscribe(val => this.next(val as T));
+    }
+
+    private performExec<T>(exec: Executor<T>, args: any) { 
+        return exec(args).pipe(
+            catchError(err => of(new Error(err)))
+        );
+    };
+
+    exec(args: any) {
+        this._executionStream.next(args);
+    }
+}
+
+export function exec<T>(exec: Executor<T>, initialValue?: T, error?: BehaviorSubject<Error>): ExecSubject<T> {
+    return new ExecSubject<T>(exec, initialValue, error);
+}
\ No newline at end of file