class Observable {
constructor(subscribe) {
this._subscribe = subscribe;
}
subscribe(observer) {
return this._subscribe(observer);
}
static of(value) {
return new Observable((observer) => {
observer.next(value);
observer.complete();
return {
unsubscribe() {}
};
});
}
static concnat(...observables) {
return new Observable((observer) => {
const copy = observables.slice();
let currentSubscription = null;
// using recurisve call to implement concat effect
// for each call, there we update currentSubscription
// when calling unsubscribe on currentSubscription
// complete() won't be triggered, therefore whole
// observables stops as well
const processObservable = () => {
if (!copy.length) {
observer.complete();
} else {
const observable = copy.shift();
currentSubscription = observable.subscribe({
next(data) {
observer.next(data);
},
error(err) {
observer.error(err);
currentSubscription.unsubscribe();
},
complete() {
processObservable();
}
});
}
};
processObservable();
return {
unsubscribe: () => {
currentSubscription.unsubscribe();
}
};
});
}
static timeout(time) {
return new Observable((observer) => {
console.log("CALL TIMEOUT");
const handle = setTimeout(() => {
observer.next();
observer.complete();
}, time);
return {
unsubscribe: () => {
clearTimeout(handle);
}
};
});
}
static fromEvent(dom, eventName) {
return new Observable((observer) => {
const handle = (e) => {
observer.next(e);
};
dom.addEventListener(eventName, handle);
return {
unsubscribe: () => {
dom.removeEventListener(eventName, handle);
}
};
});
}
retry(num) {
return new Observable((observer) => {
let currentSub;
const retry = (currentAmttemptNumber) => {
currentSub = this.subscribe({
next(data) {
observer.next(data);
},
error(err) {
if (currentAmttemptNumber === 0) {
observer.error(err);
} else {
retry(currentAmttemptNumber - 1);
}
},
complete() {
observer.complete();
}
});
};
retry(num);
return {
unsubscribe() {
if (currentSub) {
currentSub.unsubscribe();
}
}
};
});
}
// map is a operator, you need to subscribe it
map(transfomer) {
return new Observable((observer) => {
const subscription = this.subscribe({
next(data) {
let value;
try {
value = transfomer(data);
observer.next(value);
} catch (err) {
observer.error(err);
subscription.unsubscribe();
}
},
error(e) {
observer.error(e);
},
complete() {
observer.complete();
}
});
return subscription;
});
}
filter(predition) {
return new Observable((observer) => {
const subscription = this.subscribe({
next(data) {
try {
if (predition(data)) {
observer.next(data);
}
} catch (err) {
observer.error(err);
subscription.unsubscribe();
}
},
error(e) {
observer.error(e);
},
complete() {
observer.complete();
}
});
return subscription;
});
}
share() {
const subject = new Subject();
this.subscribe(subject);
return subject;
}
observeOn(scheduler) {
return new Observable((observer) => {
return this.subscribe({
next(v) {
scheduler(() => observer.next(v));
},
error(e) {
scheduler(() => observer.error(e));
},
complete() {
scheduler(() => observer.complete());
}
});
});
}
subscribeOn(scheduler) {
return new Observable((observer) => {
scheduler(() =>
this.subscribe({
next(v) {
observer.next(v);
},
error(e) {
observer.error(e);
},
complete() {
observer.complete();
}
})
);
});
}
}
class Subject extends Observable {
constructor() {
super((observer) => {
this.observers.add(observer);
return {
unsubscribe() {
this.observers.delete(observer);
}
};
});
this.observers = new Set();
}
next(v) {
for (let observer of [...this.observers]) {
observer.next(v);
}
}
error(e) {
for (let observer of [...this.observers]) {
observer.next(e);
}
}
complete() {
for (let observer of [...this.observers]) {
observer.complete();
}
}
}
Observable.of(5)
.observeOn((action) => {
setTimeout(action, 1000);
})
.subscribe({
next(v) {
console.log(v);
},
complete() {
console.log("done");
}
});
Observable.of(6)
.subscribeOn((action) => {
setTimeout(action, 1000);
})
.subscribe({
next(v) {
console.log(v);
},
complete() {
console.log("done");
}
});