Merge branch 'next'

Author: maniator

README.md

@@ -49,75 +49,10 @@ NEXT NUMBER:  50 // after a 1 second delay
 I AM COMPLETE // same time as previous line
 ```
 
-### There a few plugins that are available - each return a new Observable instance:
+### There a few plugins that are available - each return a new Observable instance
 
-- They are called by doing: `observableInstance$.<pluginFunction>`
+- [Documentation on the plugin operators can be seen here](./src/operators/DOCUMENTATION.md)
 
-`.map(<mapCallback>)` 
+### There are also a few Observables that can be created from the Observable object itself
 
- - will map each value to a new value using the callback
-    
-`.filter(<filterCallback>)` 
-
-- will filter out values that you do not want to subscribe to
-    
-`.do(<doCallback>)`
-
-- will run some callback before passing the current value to the subscription
-    
-`.take(<amount>[, <callback>])` 
-
-- will take the amount of values you want (or less if it completes first) and then complete -- this is useful for infinitely running observables
-- a callback may be passed to filter out the valeus you want to take
-    
-`.first([<callback>])` 
-
- - this is an alias for `.take(1[, <callback>])`
- 
- `.toPromise()`
- 
- - this will do a `.first()` on the observable object and return the value from that to the response from a promise
- 
- `.flatMap(<mapCallback>)`
- 
- - same as `.map(<mapCallback>)` but will take the value of the callback and turn it from an observable to a value
- 
- `.switchMap(<mapCallback>)`
- 
- - the value from the mapCallback is an observable and if another value comes through the previous observable is cancelled
-    - this is useful for things like typeaheads where you dont want a request for every keypress
-    
-`.delay(<time>)`
-
-- will delay output from the observable until a specific time interval has passed
-
-`.debounceTime(<time>)`
-
-- will only output values if there has not been any new values in the past time interval passed
-
-### There are also a few Observables that can be created from the Observable object itself:
-
-- They are called by doing: `Observable.<observableFunction>`
-
-`.fromEvent(<eventName>, <element>[, <mapCallback>)`
-
-- will create an observable that will listen for an event on a DOM element
-- there is also an optional mappCallback that can be used to map the event that comes through the next observer function
-
-`.fromPromise(<promise>)`
-
-- turns a promise into an observable that emits the value of the promise and then completes
-
-`.interval(<time>[, <start = 0>])`
-
-- emits a number at the specified time interval and increases by one every time
-- there is an optional start value that can be passed
-
-`.of(<...values>)`
-
-- takes any number of arguments and emits them in an observable (in order) and then completes
-
-`.range(<start>[, <end>])`
-
-- emits a range of numbers from start to end and then completes
-- if only start is given than the range starts at 0 and ends at the start value
+- [Documentation on the observables can be seen here](./src/observables/DOCUMENTATION.md)

example/index.js

@@ -1,24 +1,28 @@
 const { Observable } = Servable;
 
-// const countObservable$ = Observable.interval(1000, 1);
-//
-// const countSubscription = countObservable$
-//   .take(10)
-//   .map((n) => n * 5)
-//   .filter((n) => n > 10)
-//   .subscribe({
-//     next (number) {
-//       console.log('NEXT NUMBER: ', number);
-//     },
-//
-//     error (errors) {
-//       console.warn('I HAVE ERRORS', errors)
-//     },
-//
-//     complete () {
-//       console.log('I AM COMPLETE');
-//     }
-//   });
+const countObservable$ = Observable.interval(1000, 1);
+
+const countSubscription = countObservable$
+  .take(10)
+  .map((n) => n * 5)
+  .filter((n) => n > 10)
+  .combineLatest(
+    countObservable$
+      .take(5)
+  )
+  .subscribe({
+    next (number) {
+      console.log('NEXT NUMBER: ', number);
+    },
+
+    error (errors) {
+      console.warn('I HAVE ERRORS', errors)
+    },
+
+    complete () {
+      console.log('I AM COMPLETE');
+    }
+  });
 
 // test event binding
 const inputObservable$ = Observable.fromEvent('input', document.getElementById('myInput'), (event) => event.currentTarget.value);

package.json

@@ -1,6 +1,6 @@
 {
   "name": "servable",
-  "version": "0.1.5",
+  "version": "0.2.0",
   "description": "From scratch observable",
   "main": "dist/index.js",
   "scripts": {

src/Observable.js

@@ -6,6 +6,10 @@ export class Observable {
     this.observerCallback = observerCallback;
   }
 
+  static create (observerCallback) {
+    return new Observable(observerCallback);
+  }
+  
   subscribe (next = noop, error = noop, complete = noop) {
     return new Subscription(this.observerCallback, { next, error, complete });
   }

src/Observer.js

@@ -0,0 +1,81 @@
+import { noop } from './utilities';
+
+export class Observer {
+  constructor (observer = {}) {
+    this.isComplete = false;
+    this.dispose = noop;
+    
+    this.setupObserver(observer);
+  }
+  
+  cleanup () {
+    this.catchErrors(() => {
+      this.dispose();
+  
+      this.onNext = noop;
+      this.onError = noop;
+      this.onComplete = noop;
+      this.dispose = noop;
+      
+      this.isComplete = true;
+    })();
+    
+    return this;
+  }
+  
+  catchErrors (callback) {
+    return (...args) => {
+      try {
+        return callback(...args);
+      } catch (errors) {
+        this.onError(errors);
+      }
+    };
+  }
+  
+  use (callback) {
+    return this.catchErrors(() => {
+      const response = callback({
+        next: (...args) => this.onNext(...args),
+        error: (...errors) => this.onError(...errors),
+        complete: () => this.onComplete(),
+      });
+      
+      if (typeof response === 'function') {
+        this.dispose = response;
+      } else {
+        this.dispose = noop;
+      }
+      
+      return this.dispose;
+    })();
+  }
+  
+  setupObserver ({ next = noop, error = noop, complete = noop }) {
+    // assumes that an object was passed as first value to subscription
+    if (typeof next !== 'function' && typeof next === 'object') {
+      return this.setupObserver(next);
+    }
+  
+    this.onNext = this.catchErrors((...args) => {
+      if (!this.isComplete) {
+        return next(...args);
+      }
+    });
+  
+    this.onError = (...errors) => {
+      if (!this.isComplete) {
+        return error(...errors);
+      }
+    };
+  
+    this.onComplete = this.catchErrors(() => {
+      if (!this.isComplete) {
+        this.cleanup();
+        return complete();
+      }
+    });
+  
+    return this;
+  }
+}

src/Subject.js

@@ -1,49 +1,64 @@
-import { Subscription } from './Subscription';
-import { noop } from './utilities';
+import {Subscription} from './Subscription';
 
 export class Subject {
   constructor () {
     this.next = this.next.bind(this);
     this.error = this.error.bind(this);
     this.complete = this.complete.bind(this);
 
-    this.subscriptionList = [];
+    this.observers = [];
   }
 
-  subscribe (next = noop, error = noop, complete = noop) {
-    const subscription = new Subscription(noop, {next, error, complete});
+  subscribe (observer) {
+    const subscription = Subscription.createSimple(observer);
 
-    this.subscriptionList.push(subscription);
+    this.observers.push(subscription.observer);
 
     return subscription;
   }
 
+  unsubscribe () {
+    this
+      .cleanup((observer) => {
+        observer.cleanup();
+        
+        return false;
+      });
+  }
+  
   cleanup (callback) {
-    this.subscriptionList = this.subscriptionList.filter((subscription) => {
-      if (!subscription.isComplete) {
-        callback(subscription);
+    this.observers = this.observers.filter((observer) => {
+      if (!observer.isComplete) {
+        const ret = callback(observer);
+        
+        if (typeof ret === 'boolean') {
+          return ret;
+        }
+        
         return true;
       }
 
       return false;
     });
+    
+    return this;
   }
 
   next (...args) {
-    this.cleanup((subscription) => {
-        subscription.next(...args);
+    this.cleanup((observer) => {
+        observer.onNext(...args);
     });
   }
 
   error (...errors) {
-    this.cleanup((subscription) => {
-      subscription.error(...errors);
+    this.cleanup((observer) => {
+      observer.onError(...errors);
     });
   }
 
   complete () {
-    this.cleanup((subscription) => {
-      subscription.complete();
+    this.cleanup((observer) => {
+      observer.onComplete();
     });
   }
 }