Merge tag '0.3.2'

v0.3.2

Author: maniator

README.md

@@ -10,45 +10,6 @@ npm i servable --save
 
 ---------
 
-It can be used like so:
-
-```
-const Observable = require('servable').Observable;
-
-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');
-    }
-  });
-```    
-
-This will log out to the console (if subscribe like above is called):
-```
-NEXT NUMBER:  15
-NEXT NUMBER:  20 // after a 1 second delay
-NEXT NUMBER:  25 // after a 1 second delay
-NEXT NUMBER:  30 // after a 1 second delay
-NEXT NUMBER:  35 // after a 1 second delay
-NEXT NUMBER:  40 // after a 1 second delay
-NEXT NUMBER:  45 // after a 1 second delay
-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
 
 - [Documentation on the plugin operators can be seen here](./src/operators/DOCUMENTATION.md)

example/adder.js

@@ -1,6 +1,6 @@
 // same example as https://github.com/cujojs/most#simple-example
 
-const { fromEvent, combineLatest } = Servable;
+const { fromEvent, combine } = Servable;
 
 const xInput = document.querySelector('input.x');
 const yInput = document.querySelector('input.y');
@@ -22,7 +22,7 @@ const main = () => {
   const y = fromEvent('input', yInput).map(toNumber);
 
   // result is the live current value of adding x and y
-  const result = combineLatest(x, y, add);
+  const result = combine([x, y], add);
 
   // Observe the result value by rendering it to the resultNode
   result.subscribe(renderResult);

example/index.html

@@ -18,7 +18,7 @@ <h3>Adder</h3>
         </form>
     </div>
 
-    <script type="text/javascript" src="../dist/index.min.js"></script>
+    <script type="text/javascript" src="../dist/index.js"></script>
     <script type="text/javascript" src="index.js"></script>
     <script type="text/javascript" src="adder.js"></script>
 </body>

example/index.js

@@ -1,15 +1,16 @@
 const { Observable } = Servable;
 
-const countObservable$ = Observable.interval(1000, 1);
-
-const countSubscription = countObservable$
-  .take(10)
-  .map((n) => n * 5)
-  .filter((n) => n > 10)
-  .combineLatest(
-    countObservable$
-      .take(5)
-  )
+const countObservable$ = Observable.interval(1000, 1).take(4);
+const countObservable2$ = Observable.interval(500, 5);
+
+const countSubscription =
+  Observable.zip([
+    countObservable$,
+    countObservable2$.take(7),
+    countObservable$,
+    countObservable2$.take(7),
+  ])
+// countObservable2$.take(5)
   .subscribe({
     next (number) {
       console.log('NEXT NUMBER: ', number);
@@ -20,7 +21,7 @@ const countSubscription = countObservable$
     },
 
     complete () {
-      console.log('I AM COMPLETE');
+      console.trace('I AM COMPLETE');
     }
   });
 
@@ -39,3 +40,13 @@ inputObservable$
 inputObservable$
   .do((text) => div.textContent = text)
   .subscribe();
+
+Observable
+  .ajax('../package.json')
+  .do(console.log.bind(console.log, 'response'))
+  .flatMap(response => Observable.fromPromise(response.json()))
+  .do(console.log.bind(console.log, 'value'))
+  .subscribe({
+    next: (value) => console.log(value),
+    error: (e) => console.warn(e, e.response)
+  });

package.json

@@ -1,14 +1,15 @@
 {
   "name": "servable",
-  "version": "0.3.1",
+  "version": "0.3.2",
   "description": "From scratch observable",
   "main": "lib/index.js",
   "jsnext:main": "src/index.js",
   "scripts": {
     "test": "echo \"Error: no test specified\" && exit 0",
     "build": "rimraf lib dist && buba src -o lib && rollup -c && uglifyjs dist/index.js -c \"warnings=false\" --comments -m -o dist/index.min.js --verbose",
     "watch": "rollup -c -w",
-    "release": "node release.js"
+    "release": "node release.js",
+    "server": "http-server"
   },
   "repository": {
     "type": "git",
@@ -22,13 +23,15 @@
   "homepage": "https://github.com/maniator/servable#readme",
   "devDependencies": {
     "buba": "^4.0.2",
+    "http-server": "^0.10.0",
     "inquirer": "^3.0.6",
     "rimraf": "^2.6.1",
     "rollup": "^0.41.6",
     "rollup-plugin-buble": "^0.15.0",
+    "rollup-plugin-filesize": "^1.3.2",
     "rollup-plugin-license": "^0.3.0",
+    "rollup-plugin-node-globals": "^1.1.0",
     "rollup-plugin-node-resolve": "^3.0.0",
-    "rollup-plugin-sizes": "^0.3.0",
     "rollup-watch": "^3.2.2",
     "semver": "^5.3.0",
     "shelljs": "^0.7.7",

rollup.config.js

@@ -1,7 +1,8 @@
 import buble from 'rollup-plugin-buble';
 import nodeResolve from 'rollup-plugin-node-resolve';
-import sizes from 'rollup-plugin-sizes';
+import filesize from 'rollup-plugin-filesize';
 import license from 'rollup-plugin-license';
+import globals from 'rollup-plugin-node-globals';
 
 const pkg = require('./package.json');
 
@@ -18,7 +19,8 @@ export default {
   sourceMap: true,
   plugins: [
     buble(),
-    sizes(),
+    filesize(),
+    globals(),
     license({
       banner,
     }),

src/Observer.js

@@ -36,7 +36,7 @@ export class Observer {
   use (callback) {
     return this.catchErrors(() => {
       const response = callback({
-        next: (...args) => this.onNext(...args),
+        next: (value) => this.onNext(value),
         error: (...errors) => this.onError(...errors),
         complete: () => this.onComplete(),
       });
@@ -57,9 +57,9 @@ export class Observer {
       return this.setupObserver(next);
     }
 
-    this.onNext = this.catchErrors((...args) => {
+    this.onNext = this.catchErrors((value) => {
       if (!this.isComplete) {
-        return next(...args);
+        return next(value);
       }
     });
 

src/Subject.js

@@ -44,9 +44,9 @@ export class Subject {
     return this;
   }
 
-  next (...args) {
+  next (value) {
     this.cleanup((observer) => {
-        observer.onNext(...args);
+        observer.onNext(value);
     });
   }
 

src/observables.js

@@ -1,3 +1,4 @@
+export * from './observables/ajax';
 export * from './observables/fromEvent';
 export * from './observables/fromPromise';
 export * from './observables/range';

src/observables/DOCUMENTATION.md

@@ -1,5 +1,10 @@
 These methods are called by doing: `Observable.<observableFunction>`
 
+`.ajax(url[, options])`
+
+- an observable wrapper around the `fetch` API
+    - you can add a shim like [this one](https://github.com/github/fetch) if your environment does not have the fetch api
+
 `.fromEvent(<eventName>, <element>[, <mapCallback>)`
 
 - will create an observable that will listen for an event on a DOM element

src/observables/ajax.js

@@ -0,0 +1,29 @@
+import { Observable } from '../Observable';
+import { fromPromise } from './fromPromise';
+
+function checkStatus (response) {
+  if (response.status >= 200 && response.status < 300) {
+    return response;
+  } else {
+    var error = new Error(response.statusText);
+    error.response = response;
+    throw error;
+  }
+}
+
+// do an ajax call (depends on the "fetch" api existing)
+// can add this shim to simulate fetch in all browsers https://github.com/github/fetch
+export const ajax = function (...args) {
+  return new Observable(function (observer) {
+    if (global.fetch) {
+      const fetchPromise = fetch(...args).then(checkStatus);
+      const subscription = fromPromise(fetchPromise).subscribe(observer);
+  
+      return () => subscription.unsubscribe();
+    } else {
+      throw new Error('Fetch API does not exist in your environment');
+    }
+  });
+};
+
+Observable.ajax = ajax;

src/observables/fromPromise.js

@@ -4,10 +4,10 @@ import { isPromise } from '../utilities';
 export const fromPromise = function (promise) {
   return new Observable(function ({ next, complete, error }) {
     if (isPromise(promise)) {
-      promise.then(function (...values) {
-        next(...values);
+      promise.then(function (values) {
+        next(values);
         complete();
-      }, error);
+      }, error).catch(error);
     } else {
       error('Passed an invalid object to fromPromise', promise);
     }

src/operators.js

@@ -8,10 +8,12 @@ export * from './operators/first';
 export * from './operators/toPromise';
 export * from './operators/flatMap';
 export * from './operators/switchMap';
+export * from './operators/merge';
 export * from './operators/scan';
 export * from './operators/reduce';
 export * from './operators/concat';
-export * from './operators/combineLatest';
+export * from './operators/zip';
+export * from './operators/combine';
 export * from './operators/count';
 export * from './operators/sum';
 export * from './operators/max';

src/operators/DOCUMENTATION.md

@@ -1,4 +1,4 @@
-These methods called by doing: `observableInstance$.<pluginFunction>`
+These methods called by doing: `observableInstance$.<pluginFunction>` or by calling `Observable.<pluginFunction>(observableInstance$, ...arguments)`
 
 `.map(<mapCallback>)` 
 
@@ -75,7 +75,7 @@ These methods called by doing: `observableInstance$.<pluginFunction>`
 
 - will run any number of observables in order and will only complete when the last one is done
 
-`.combineLatest(otherSource$[, combineCallback])`
+`.combine(otherSources$[, combineCallback])`
 
-- will run two observables at the same time
-    - will only call the observer's next function when both observables have emitted at least one value 
+- will multiple observables at the same time
+    - will only call the observer's next function when all observables have emitted at least one value 

src/operators/average.js

@@ -21,6 +21,7 @@ export const average = function (source$) {
   });
 };
 
+Observable.average = average;
 Observable.prototype.average = function () {
   return average(this);
 };

src/operators/combine.js

@@ -0,0 +1,41 @@
+import { Observable } from '../Observable';
+import { onSubscriptionsComplete } from '../utilities/onSubscriptionsComplete';
+
+const nullHash = void(0);
+
+export const combine = function (sources$, combineCallback = ((...args) => [...args])) {
+  return new Observable(function ({ next, error, complete }) {
+    let subscriptions = [];
+    
+    let latest = sources$.map(s$ => nullHash);
+    
+    let allHasValue = false;
+    const checkAllHasValue = () => latest.filter((l) => l == nullHash).length <= 0;
+    
+    const onComplete = () => onSubscriptionsComplete(subscriptions, complete);
+    const subscribeTo = (obs$, index) => {
+      return obs$.subscribe({
+        next (value) {
+          latest[index] = value;
+  
+          allHasValue = allHasValue || checkAllHasValue();
+          
+          if (allHasValue) {
+            next(combineCallback(...latest));
+          }
+        },
+        error,
+        complete: onComplete,
+      });
+    };
+  
+    subscriptions = sources$.map((s$, index) => subscribeTo(s$, index));
+    
+    return () => subscriptions.forEach((s) => s.unsubscribe());
+  });
+};
+
+Observable.combine = combine;
+Observable.prototype.combine = function (otherSources$, combineCallback) {
+  return combine([this, ...otherSources$], combineCallback);
+};