Promise-observer NPM

promise-observer

An observer implementation with promise support.

Its meant to behave similarly to typical synchronous Java / NET observers, where once you notify your subscribers you are able to wait for their update functions to execute before proceeding

Example

Given a blog post creator:

import po = require('promise-observer')
function BlogPostCreator() {
    this.onCreated = po.create(emit => this.emit = emit);
}
BlogPostCreator.prototype.create = function(blogPost) {
    actuallyCreateBlogpost()
        .then(post => this.emit(post))
        // wait for all attached events to complete before commiting.
        .then(commitTransaction);
}

a categorizer can attach to its events

onPostCategorized = blogPostCreator.onCreated(post =>
  categorize(post).then(saveCategory).thenReturn(post));

an indexer can add search terms to the index for that post

onPostIndexed = blogPostCreator.onCreated(post =>
  index(post).then(saveIndex).thenReturn(post));

Then, the email notification system can wait for the post to be categorized and indexed before sending a notification to all subscribers:

onPostNotification = blogPostCreator.onCreated(post => {
  var categorized = onPostCategorized.next(categorizedPost => categorizedPost.id == post.id);
  var indexed = onPostIndexed.next(indexedPost => indexedPost.id == post.id);
  return Promise.join(categorized, indexed, _ => sendEmailNotification(post))
});

API

po.create(emit):Observable

po.create(emit: (val:T) => Promise<void>):Observable<T>

Creates a new observable. The observable exposes its emit function through the revealing constructor pattern. Use the emit function to notify all subscribers of new events.

The emit function returns a promise that resolves when all subscribers and their dependents finish processing the event.

Observable

interface Observable<T> {
    <U>(listener: (t: T) => U): LinkedObservable<U>;
    <U>(listener: (t: T) => Promise<U>): LinkedObservable<U>;
    next(predicate?: (t: T) => boolean): Promise<T>;
    remove<U>(o: Observable<U>): void;
}