Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

disposing an observable should dispose its rxObservable #6

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

disposing an observable should dispose its rxObservable #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 23 additions & 3 deletions out/knockout.rx.js
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion out/knockout.rx.js.map
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

199 changes: 109 additions & 90 deletions src/knockout.rx.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,95 +9,114 @@ declare var exports: any;
declare var module: any;

(function (factory: (rx: typeof Rx, ko: KnockoutStatic) => void) {
if (typeof require === "function" && typeof exports === "object" && typeof module === "object") {
factory(require("rx"), require("knockout"));
} else if (typeof define === "function" && define["amd"]) {
define(["rx", "knockout"], factory);
} else {
factory(Rx, ko);
}
if (typeof require === "function" && typeof exports === "object" && typeof module === "object") {
factory(require("rx"), require("knockout"));
} else if (typeof define === "function" && define["amd"]) {
define(["rx", "knockout"], factory);
} else {
factory(Rx, ko);
}
}(function(rx: typeof Rx, ko: KnockoutStatic) {
var rxObservableProto = <Rx.Observable<any>>(<any>rx.Observable).prototype;

function noop() {}

function ko2rx<T>(event?: string): Rx.Observable<T> {
// KnockoutSubscription implements Rx._IDisposable
return rx.Observable.createWithDisposable<T>(observer => (<KnockoutSubscribable<T>>this).subscribe(observer.onNext, observer, event));
}

function ko2rxReply<T>(event?: string): Rx.Observable<T> {
return rx.Observable.createWithDisposable<T>(observer => {
observer.onNext((<KnockoutObservable<T>>this).peek());
return (<KnockoutSubscribable<T>>this).subscribe(observer.onNext, observer, event); // KnockoutSubscription implements Rx._IDisposable
});
}

function ko2rxSubject<T>(): Rx.ISubject<T> {
var observable = <KnockoutObservable<T>>this;
return rx.Subject.create<T>(
rx.Observer.create<T>(value => observable(value)),
observable.toObservable());
}

function rx2koSubscribable<T>(): KnockoutSubscribable<T> {
var subscribable = new ko.subscribable<T>();
var subscriptionCount = 0;

subscribable.subscribe = (callback: (newValue: T) => void, target?: any, event?: string): KnockoutSubscription => {

var observer = (!event || event == "change" || event == "onNext") ? Rx.Observer.create<T>(callback)
: (event == "onError" || event == "error") ? Rx.Observer.create<T>(noop, callback)
: (event == "onCompleted" || event == "complete") ? Rx.Observer.create<T>(noop, noop, <() => void>callback)
: null;

if (!observer)
throw new Error("Unknown event '" + event + "'");

subscriptionCount++;
var disposable = (<Rx.Observable<T>>this).subscribe(observer) || Rx.Disposable.empty;
return {
dispose: () => {
if (disposable) {
disposable.dispose();
disposable = null;
subscriptionCount--;
}
}
};
};

subscribable.getSubscriptionsCount = () => subscriptionCount;

return subscribable;
}

function rx2koObservable<T>(initialValue?: T): KnockoutObservable<T> {
var observable = ko.observable(initialValue);
(<Rx.Observable<T>>this).subscribe(observable);
return observable;
}

function rxSubject2koObservable<T>(initialValue?: T): KnockoutObservable<T> {
var subject = <Rx.Subject<T>>this;
var observable = ko.observable(initialValue);
var changingBySubject = false;

observable.subscribe(value => !changingBySubject && subject.onNext(value));
subject.subscribe(value => {
changingBySubject = true;
observable(value);
changingBySubject = false;
});

return observable;
}

ko.subscribable.fn.toObservable = ko2rx;
ko.observable.fn.toObservableWithReplyLatest = ko2rxReply;
ko.observable.fn.toSubject = ko2rxSubject;
ko.computed.fn.toObservableWithReplyLatest = ko2rxReply;
rxObservableProto.toKoSubscribable = rx2koSubscribable;
rxObservableProto.toKoObservable = rx2koObservable;
rx.Subject.prototype.toKoObservable = rxSubject2koObservable;
var rxObservableProto = <Rx.Observable<any>>(<any>rx.Observable).prototype;

function noop() {}

function ko2rx<T>(event?: string): Rx.Observable<T> {
// KnockoutSubscription implements Rx._IDisposable
return rx.Observable.createWithDisposable<T>(observer => (<KnockoutSubscribable<T>>this).subscribe(observer.onNext, observer, event));
}

function ko2rxReply<T>(event?: string): Rx.Observable<T> {
return rx.Observable.createWithDisposable<T>(observer => {
observer.onNext((<KnockoutObservable<T>>this).peek());
return (<KnockoutSubscribable<T>>this).subscribe(observer.onNext, observer, event); // KnockoutSubscription implements Rx._IDisposable
});
}

function ko2rxSubject<T>(): Rx.ISubject<T> {
var observable = <KnockoutObservable<T>>this;
return rx.Subject.create<T>(
rx.Observer.create<T>(value => observable(value)),
observable.toObservable());
}

function rx2koSubscribable<T>(): KnockoutSubscribable<T> {
var subscribable = new ko.subscribable<T>();
var subscriptionCount = 0;

subscribable.subscribe = (callback: (newValue: T) => void, target?: any, event?: string): KnockoutSubscription => {

var observer = (!event || event == "change" || event == "onNext") ? Rx.Observer.create<T>(callback)
: (event == "onError" || event == "error") ? Rx.Observer.create<T>(noop, callback)
: (event == "onCompleted" || event == "complete") ? Rx.Observer.create<T>(noop, noop, <() => void>callback)
: null;

if (!observer)
throw new Error("Unknown event '" + event + "'");

subscriptionCount++;
var disposable = (<Rx.Observable<T>>this).subscribe(observer) || Rx.Disposable.empty;
return {
dispose: () => {
if (disposable) {
disposable.dispose();
disposable = null;
subscriptionCount--;
}
}
};
};

subscribable.getSubscriptionsCount = () => subscriptionCount;

return subscribable;
}
function addDispose(observable, dispose) {
var subscriptions = [];
var oldSubscribe = observable.subscribe.bind(observable);
observable.subscribe = function (callback, callbackTarget, event) {
var oldDispose, ret;
ret = oldSubscribe(callback, callbackTarget, event);

subscriptions.push(ret);
oldDispose = ret.dispose;
ret.dispose = function () {
subscriptions.splice(subscriptions.indexOf(ret), 1);
oldDispose.bind(this)();
if (!subscriptions.length) {
dispose();
}
};
return ret;
};
}
function rx2koObservable<T>(initialValue?: T): KnockoutObservable<T> {
var observable = ko.observable(initialValue);
var rxSubscription = (<Rx.Observable<T>>this).subscribe(observable);
addDispose(observable, ()=> rxSubscription.dispose() );
return observable;

}

function rxSubject2koObservable<T>(initialValue?: T): KnockoutObservable<T> {
var subject = <Rx.Subject<T>>this;
var observable = ko.observable(initialValue);
var changingBySubject = false;

observable.subscribe(value => !changingBySubject && subject.onNext(value));
subject.subscribe(value => {
changingBySubject = true;
observable(value);
changingBySubject = false;
});
return observable;
}

ko.subscribable.fn.toObservable = ko2rx;
ko.observable.fn.toObservableWithReplyLatest = ko2rxReply;
ko.observable.fn.toSubject = ko2rxSubject;
ko.computed.fn.toObservableWithReplyLatest = ko2rxReply;
rxObservableProto.toKoSubscribable = rx2koSubscribable;
rxObservableProto.toKoObservable = rx2koObservable;
rx.Subject.prototype.toKoObservable = rxSubject2koObservable;
}));
37 changes: 37 additions & 0 deletions tests/knockout.rx-tests.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ test("Rx.IObservable.toKoObservable(initialValue)", () => {
equal(xo(), 1, "changed value");
});


test("Rx.IObservable.toKoSubscribable()", () => {
var xs = new Rx.Subject<number>();
var xo = xs.toKoSubscribable();
Expand Down Expand Up @@ -108,6 +109,42 @@ test("Rx.Subject.toKoObservable", () => {
"notified after ko's set");
});



test("Rx.Observable.toKoObservable() disposal", () => {
var isDisposed;
var xs = Rx.Observable.create((observer) => () => isDisposed = true);

var xo = xs.toKoObservable(20);

var disposer = xo.subscribe(function () { });
//var disposer2 = xo.subscribe(function () { });
disposer.dispose();


equal(isDisposed, true);

});


test("Rx.Observable.toKoObservable() two subscribers disposal", () => {
var isDisposed;
var xs = Rx.Observable.create((observer) => () => isDisposed = true);

var xo = xs.toKoObservable(20);

var disposer = xo.subscribe(function () { });
var disposer2 = xo.subscribe(function () { });
disposer.dispose();

equal(isDisposed, undefined);

disposer2.dispose();

equal(isDisposed, true);

});

test("ko.observable.toObservable()", () => {
var xo = ko.observable(1);
var xs = xo.toObservable();
Expand Down