ObservableはESと干渉しそうなので、仮名としてGaobservableにしました。

00.Gaobservable.js

"use strict";

const resolverSymbol     = Symbol("resolver");
const notifierListSymbol = Symbol("notifierList");
const cancelSymbol       = Symbol("cancel");

class Gaobservable{
    constructor(resolver){
        if(typeof(resolver) !== "function") throw new TypeError(`Gaobservable resolver ${resolver} is not a function`);
        
        this[resolverSymbol] = resolver;
        this[notifierListSymbol] = [];
    }
    
    tap(notifier){
        if(typeof(notifier) !== "function") return this;
        
        this[notifierListSymbol].push(notifier);
        
        return this;
    }
    
    toPromise(){
        const resolver     = this[resolverSymbol];
        const notifierList = this[notifierListSymbol];
        
        const promise = new Promise(function executor(promiseResolve, promiseReject){
            const data = [];
            let finalizer;
            
            try{
                finalizer = resolver(resolve, reject, add, notify);
            }
            catch(err){
                reject(err);
            }
            
            function resolve(){
                if(typeof(finalizer) === "function"){
                    try{
                        finalizer();
                    }
                    catch(err){
                        finalizer = null;
                        reject(err);
                        return;
                    }
                }
                
                promiseResolve(data);
            }
            
            function reject(reason){
                if(typeof(finalizer) === "function"){
                    try{
                        finalizer();
                    }
                    catch(err){
                        finalizer = null;
                        reject(err);
                        return;
                    }
                }
                
                promiseReject(reason);
            }
            
            function add(datum){
                notify(datum);
                data.push(datum);
            }
            
            function notify(datum){
                try{
                    for(var i=0,iLen=notifierList.length;i<iLen;i++){
                        notifierList[i](resolve, reject, datum);
                    }
                }
                catch(err){
                    reject(err);
                }
            }
        });
        
        return promise;
    }
    
    toCancelable(){
        const promise   = this.toPromise();
        let isCanceled = false;
        let cancelReason, reject;
        
        function cancel(reason){
            if(!isCanceled){
                isCanceled   = true;
                cancelReason = reason;
            }
            
            if(reject) reject({
                token: Gaobservable.cancelToken,
                reason: cancelReason,
            });
        }
        
        const cancelPromise = new Promise((_, promiseReject)=> {
            reject = promiseReject;
            if(isCanceled) cancel();
        });
        
        return [Promise.race([promise, cancelPromise]), cancel];
    }
    
    static create(resolver, notifier){
        return new Gaobservable(resolver)
            .tap(notifier)
            .toPromise();
    }
    
    static get cancelToken(){ return cancelSymbol; }
}

export default Gaobservable

01.sample.js

import Gaobservable from "./Gaobservable"

Promise.resolve("begin")
    .then((data)=> new Gaobservable((resolve, reject, add, notify)=> {
            console.log(data); // "begin"
            
            let cnt = 10;
            const timerId = setInterval(()=>{
                add(cnt--);
            }, 100);
            
            return ()=> {
                clearInterval(timerId);
                console.log("finalized");
            };
        })
        .tap((resolve, reject, data)=> {
            console.log("tapped:", data);
            if(!data) resolve();
        })
        .toPromise())
    .then((data)=> {
        return data.reduce((a,b)=> a+b);
    })
    .then((data)=> {
        console.log(data); // 55
    })
    .then(()=> Gaobservable.create((resolve, reject, add, notify)=> {
        throw new Error("ぬるぽ");
    }))
    .catch((err)=> {
        console.log("error:", err.message); // "error: ぬるぽ"
    });

/* 実行結果
begin
tapped: 10
tapped: 9
tapped: 8
tapped: 7
tapped: 6
tapped: 5
tapped: 4
tapped: 3
tapped: 2
tapped: 1
tapped: 0
finalized
55
error: ぬるぽ
*/

02.sampleのrepl.md

Babel repl