Repository navigation
Conversation
Member
I think the As for providing a library, which we don't to for react, we can go the https://github.com/exercism/javascript/blob/master/scripts/helpers.js#L77-L80 I worked on a somewhat spec compliant observable thing in ~350 bytes, which we could include? All I would need to do (or you, if you find it interesting) is rolling the output up into a single file. /* eslint-disable @typescript-eslint/no-non-null-assertion */
/**
* Observable provides an interface for an event system which can emit multiple values,
* is lazily evaluated and can be unsubscribed from. The laws of observable
* are:
*
* - Observable is not async, but can be easily used in an async way
* - Observable puts composability above all else (see utils below)
* - Observable is not evaluated until someone subscribes
* - Observable is no longer evaluated after someone unsubscribes
*
* @example <caption>Event listener as observable, emitting events</caption>
*
* mouseClicks = new Observable(({next} => {
* window.addEventListener('click', next)
* return () => { window.removeEventListener('click', next) }
* })
*
* let unsubscribe = mouseClicks.subscribe(({ preventDefault }) => preventDefault())
*/
class Observable {
constructor(subscribe) {
// subscribe wraps the <subscriberCallback> which only accepts proper
// <subscription> objects, and instead provides a subscribe function which
// accepts <sloppySubscription>, allowing for omitting properties in a
// subscriber object, without causing failures in the subscriberCallback
this.subscribe = (sloppyObserver, error, complete) => {
let start;
let next;
if (typeof sloppyObserver === 'function') {
next = sloppyObserver;
}
else {
start = sloppyObserver.start;
next = sloppyObserver.next;
error = sloppyObserver.error;
complete = sloppyObserver.complete;
}
let cleanup;
let closed = false;
const wrapClosed = (fn) => (v) => closed || (fn && fn(v));
const unsubscribe = wrapClosed(() => {
closed = true;
cleanup && cleanup();
});
const wrapUnsubscribe = (fn) => (v) => {
unsubscribe();
typeof fn === 'function' && fn(v);
};
error = wrapClosed(wrapUnsubscribe(error));
const wrapTry = (fn) => (v) => {
try {
fn(v);
}
catch (e) {
typeof error === 'function' && error(e);
}
};
complete = wrapClosed(wrapTry(wrapUnsubscribe(complete)));
next = wrapClosed(wrapTry(next));
const subscription = {
get closed() {
return closed;
},
unsubscribe
};
start && start(subscription);
if (closed) {
return subscription;
}
wrapTry(() => {
const observer = {
closed,
complete: complete,
error: error,
next: next
};
const wSub = subscribe(observer);
cleanup = wSub;
if (wSub && typeof wSub.unsubscribe === 'function') {
cleanup = wSub.unsubscribe;
}
})();
return subscription;
};
}
}
/**
* Creates an Observable (`Observable<T>`) from a list of items in an Array or Array Like (`Array<T>`).
*
* @param {T[]} items the source items
* @returns {ObservableT<T>} the observable that emmits the items
*/
function of(...items) {
return new Observable(({ next, complete }) => {
for (const item of items) {
next(item);
}
complete();
});
}
/**
* Creates an observable that combines two observables via a transform function
*
* {@link combine} takes two Observables (`Observable<T>` and `Observable<U>`) and when it has received values from both
* Observables will call `transform(T, U)`. {@link combine} itself returns an Observable (`Observable<V>`) which emits
* `next(V)` for every return value of the called `transform(T, U)`. `transform` can be called with stale values, if -
* for example - `sourceB` emits after `sourceA` completes, then transform will be called with the last value from
* `sourceA`. The output `Observable<V>` will only {@link complete()} when both sources {@link complete()}.
* Unsubscribing from `Observable<V>` will unsubscribe from all sources.
*
* @example <caption>Marble diagram with synchronous sources</caption>
*
* // transform t: (a, b) => a + b
* // source a: |--1--2--1--2--|
* // source b: |--1--1--2--2--|
* // combine(t, a, b): |--2--3--3--4--|
*
* @example <caption>Marble diagram with asynchronous sources</caption>
*
* // transform t: (a, b) => a + b
* // source a: |--1-----2-----3--|
* // source b: |--1--2--3--4--5--|
* // combine(t, a, b): |--2--3--5--6--8--|
*
* @param {ObservableT<T>} sourceA the first source
* @param {ObservableT<U>} sourceB the second source
* @param {((a: T, b: U) => V)} transform function to map from the two sources to the new value
*
* @returns {ObservableT<V>} the observable that has the combined values
*/
function combine(sourceA, sourceB, transform) {
return new Observable(({ error, next, complete }) => {
let sourceAComplete = false;
let sourceBComplete = false;
let sourceAStarted;
let sourceBStarted;
let sourceAValue;
let sourceBValue;
sourceA.subscribe({
error,
complete() {
sourceAComplete = true;
sourceBComplete && complete();
},
next(value) {
sourceAValue = value;
sourceAStarted = true;
sourceBStarted && next(transform(sourceAValue, sourceBValue));
}
});
sourceB.subscribe({
error,
complete() {
sourceBComplete = true;
sourceAComplete && complete();
},
next(value) {
sourceBValue = value;
sourceBStarted = true;
sourceAStarted && next(transform(sourceAValue, sourceBValue));
}
});
});
}
/**
* Combine an object of observables into a single observable which outputs an object.
*
* {@link combineObject} takes an Object of Observables and returns an Observable of objects; the keys of which match
* the key of the sources object, and the values match the values emitted by the Observable values in the sources
* object. The returned Observable may emit stale values, if - for example - one of the sources completes, then
* subsequent objects will include the last value of that Observable. The output Observable will only {@link complete()}
* when all sources {@link complete()}. Unsubscribing from the output Observable will unsubscribe from all sources.
*
* @example <caption>Marble diagram combining two sources</caption>
*
* // sources o.a: |-1----------------2----------3--------------------------|
* // sources o.b: |-------1---------------------2-------------------3------|
* // combineObject(o): |---{a:1,b:1}--{a:2,b:1}--{a:3,b:1}{a:3,b:2}--{a:3,b:3}--|
*
* @param {{ [name: string]: ObservableT<T> }} sources
* @returns {ObservableT<Record<keyof S, T>>} the combined observable
*/
function combineObject(sources) {
return new Observable(({ error, next, complete }) => {
const total = Object.keys(sources).length;
const started = {};
let completed = 0;
const values = {};
for (const name in sources) {
if (sources.hasOwnProperty(name)) {
sources[name].subscribe({
error,
complete() {
completed += 1;
completed === total && complete();
},
next(value) {
started[name] = true;
values[name] = value;
Object.keys(started).length === total &&
next(values);
}
});
}
}
});
}
/**
* Creates an observable whos emits are debounced.
*
* {@link debounce} takes a `duration` of milliseconds, and an `Observable<T>`. Any values the source `Observable<T>`
* emits will not be emitted on the output `Observable<T>` until the duration has passed. If the source `Observable<T>`
* emits multiple values during one `duration`, then older values are discarded - in other words the output
* `Observable<T>` will only emit at-most-once per duration, with the latest value from the source `Observable<T>`.
*
* @example <caption>Marble diagram for a debounce of 200 ms</caption>
*
* // source: |--1-2-3-----4-5------6-------7----|
* // debounce(200): |---------3-------5-----6-------7--|
*
* @param {ObservableT<T>} source source observable that will be debounced
* @param {number} duration number of milliseconds it should be debounced
* @returns {ObservableT<T>} observable that debounces source emits by duration ms
*/
function debounce(source, duration) {
return new Observable(({ complete, error, next }) => {
let timer;
source.subscribe({
error,
complete: () => setTimeout(complete, duration),
next: (value) => {
clearTimeout(timer);
timer = setTimeout(next, duration, value);
}
});
});
}
/**
* Filter an observable based on a predicate
*
* {@link filter} will execute `predicate<T>` for every value emitted from `Observable<T>`. If `predicate<T>` returns
* `false`, then the output `Observable<T>` won't emit that `T`. If `predicate<T>` returns `true` then the ouput
* `Observable<T>` will emit that `T`.
*
* @example <caption>Marble diagram with a filter</caption>
*
* // predicate p: x => x % 2 == 0
* // source s: |--1--2--3--4--5--6--|
* // filter(p, s): |-----2-----4-----6--|
*
* @param {ObservableT<T>} source source observable that will be filtered
* @param {(item: T) => boolean} predicate the filter predicate
* @returns {ObservableT<T>} filtered source observable by predicate
*/
function filter(source, predicate) {
return new Observable(({ complete, error, next }) => source.subscribe({
complete,
error,
next: (value) => predicate(value) && next(value)
}));
}
// transform t: x => Observable.of(x+1, x+2))
// source s: |-1-------2-------3-------7-----|
// source t(1): |-2--3--|
// source t(2): |-3--4--|
// source t(3): |-4--5--|
// source t(7): |-8--9--|
// flatMap(t, s): |-2--3----3--4----4--5----8--9--|
/**
* Creates an observable that emits values from the observables given by transform. The source observable emits to
* transform.
*
* {@link flatMap} takes a source `Observable<T>` and passes each value emitted to `transform(T)`. The output
* `Observable<U>` of `transform(T)` is then immediately subscribed to and any values are emitted on the output
* `Observable<U>`. Sources from `transform(T)` may overlap. The output `Observable<U>` will only {@link complete()}
* when all sources from `transform(T)` also {@link complete()}. Unsubscribing from the output Observable will
* unsubscribe from all sources, including `Observable<T>` - and `transform(T)` will no longer be called.
*
* @param {ObservableT<T>} source source observable that emits to transform
* @param {(item: T) => ObservableT<U>} transform function that generates observables based on emits from source
* @returns {ObservableT<U>} the observable that emits everything from the transform return value observables
*/
function flatMap(source, transform) {
let observables = 0;
const subscriptions = [];
return new Observable(({ error, next, complete }) => {
subscriptions.push(source.subscribe({
error,
next: (value) => {
observables += 1;
subscriptions.push(transform(value).subscribe({
error,
next,
complete: () => (observables -= 1) === 0 && complete()
}));
}
}));
return () => subscriptions.forEach(({ unsubscribe }) => unsubscribe());
});
}
function fromCallback(func) {
// eslint-disable-next-line @typescript-eslint/explicit-function-return-type
return (...args) =>
// eslint-disable-next-line @typescript-eslint/explicit-function-return-type
new Observable(({ next }) => func(...args.concat(next)));
}
/**
* Return an Observable that starts listening on subscrib and stops listening on unsubscribe.
*
* {@link fromEvent} takes a name, element and optional options object. The output `Observable<Event|CustomEvent>` will
* call `element.addEventListener(name, next, options)` - thereby emitting any events from the listener, to the
* Observable. When unsubscribed, `Observable<Event|CustomEvent>` will cleanup the event listener.
*
* @param {HTMLElementLike} element the html element to subscribe on
* @param {string} name the event name (e.g. click, change, keydown)
* @param {boolean|AddEventListenerOptions} options the options to pass to the listener
*
* @returns {ObservableT<Event|CustomEvent>} the observable
*/
function fromEvent(element, name, options) {
return new Observable(({ next }) => {
element.addEventListener(name, next, options);
return () => element.removeEventListener(name, next, options);
});
}
/**
* Creates an observable that emits when the promise resolves
*
* {@link fromPromise} takes a `Promise<T>` and will emit `T` when `Promise<T>` resolves. If `Promise<T>` rejects, then
* `Observable<T>` will {@link error()}. {@link complete()} is called after `Promise<T>` resolves and `Observable<T>`
* emits - as such `Observable<T>` will only ever emit one value.
*
* @param {Promise<T>} promise the source promise
* @returns {ObservableT<T>} the observable that emits when the promise resolves
*/
function fromPromise(promise) {
return new Observable(({ next, error, complete }) => {
promise.then(next, error).then(complete);
});
}
/**
* Creates an observable that maps each value from the source via a function.
*
* {@link map} takes a source `Observable<T>` and calls `transform(T)` for every emitted value. The returned
* `Observable<U>` will emit the returned values from `transform<T>`. If transform is not callable, and is instead `U`
* then the raw value is simply used instead.
*
* @example <caption>Marble diagram of mapping an obversable</caption>
*
* // transform t: x => x * 2
* // source s: |--1--2--3--4--|
* // map(t, s): |--2--4--6--8--|
*
* @param {ObservableT<T>} source the source observable
* @param {U | transformFn<T, U>} transform the function to transform items
*
* @returns {ObservableT<U>} Transformed observable
*/
function map(source, transform) {
return new Observable(({ complete, error, next }) => {
return source.subscribe({
complete,
error,
next: (value) => next(typeof transform === 'function'
? transform(value)
: transform)
});
});
}
function merge(...sources) {
return new Observable(({ error, next, complete }) => {
let remaining = sources.length;
const subscriptions = sources.map((source) => source.subscribe({
error,
next,
complete: () => (remaining -= 1) === 0 && complete()
}));
return () => {
for (const { unsubscribe } of subscriptions) {
unsubscribe();
}
};
});
}
/**
* Create an observable that skips the first {@link count} values.
*
* {@link skip} takes a count, and a source `Observable<T>`. The returned `Observable<T>` will only emit `T`s after
* the source `Observable<T>` has emitted count times. Any values the source `Observable<T>` emits before count is
* reached will be discarded.
*
* @example <caption>Marble diagram skipping 3 values</caption>
*
* // source s: |--1--2--3--4--5--6--7--|
* // skip(s, 3): |-----------4--5--6--7--|
*
* @param {ObservableT<T>} source the source observable
* @param {number} count positive count to skip
*
* @returns {ObservableT<T>} the new observable that skips the first count values
*/
function skip(source, count) {
return new Observable(({ complete, error, next }) => {
return source.subscribe({
complete,
error,
next: (value) => {
if (count === 0) {
return next(value);
}
count -= 1;
}
});
});
}
/**
* Create an observable that skips values if they're the same as the previous one.
*
* {@link skipRepeats} takes source `Observable<T>` and returns an output `Observable<T>` which will ignore repeat
* emissions from the source. Any values emitted multiple times from the source `Observable<T>` will be discarded after
* the first emission.
*
* @note In some libraries this is called `distinctUntilChanged`.
*
* @example <caption>Marble diagram</caption>
*
* // source s: |--1--1--2--3--4--4--4--4--5-|
* // skipRepeats(s): |--1-----2--3--4-----------5-|
*
* @param {ObservableT<T>} source source observable
* @returns {ObservableT<T>} the new observable that skips repeated values
*/
function skipRepeats(source) {
return new Observable(({ complete, error, next }) => {
let lastValue = {};
return source.subscribe({
complete,
error,
next: (value) => lastValue !== value && next((lastValue = value))
});
});
}
function startWith(source, start) {
return new Observable(({ error, next, complete }) => {
next(start);
return source.subscribe({ error, next, complete });
});
}
/**
* Creates an observable that works like {@link flatMap} but unsubscribes from previous {@link transform} outputs.
*
* {@link switchLatest} takes a source `Observable<T>` and calls `transform(T)` for each emitted value. Like
* {@link flatMap}, {@link switchLatest} will immediately subscribe to any `Observable<U>` coming from `transform(T)`,
* but in addition to this, will {@link unsubscribe()} from any prior `Observable<U>`s - so that there is only ever one
* `Observable<U>` subscribed at any one time.
*
* @note In some libraries this is called `switchMap`.
*
* @example <caption>Marble diagram of various sources</caption>
*
* // transform t: x => Observable.of(x+1, x+1, x+1))
* // source: |--1-----2--3--4--------|
* // source t(1): |--2--2--2--------------| // transform returns a new one
* // source t(2): |--------3--3--3--------| // transform returns a new one, t(1) observable has been closed
* // source t(3): |-----------4--4--4-----| // transform returns a new one, t(2) observable has been closed
* // source t(4): |--------------5--5--5--| // transform returns a new one, t(3) observable has been closed
* // switchLatest: |--2--2--3--4--5--5--5--| // output observable
*
* @param {ObservableT<T>} source the source observable
* @param {(item: T) => ObservableT<U>} transform function that returns an observable based on an item
*
* @returns {ObservableT<U>} new observable that emits items coming from the latest transformed observable
*/
function switchLatest(source, transform) {
let remaining = 0;
return new Observable(({ error, next, complete }) => {
let oldSubscription = null;
const { unsubscribe } = source.subscribe({
error,
next: (value) => {
remaining += 1;
if (oldSubscription) {
remaining -= 1;
oldSubscription.unsubscribe();
}
oldSubscription = transform(value).subscribe({
error,
next,
complete: () => {
remaining -= 1;
if (remaining === 0) {
complete();
unsubscribe && unsubscribe();
}
}
});
}
});
});
}
/**
* Creates an observable that only emits values as long as {@link toggler}'s last output is `true`
*
* {@link toggle} takes a source `Observable<T>` and a toggler `Observable<boolean>`. The output `Observable<T>`
* will only ever emit `T`s while `Observable<boolean>`'s last emitted value was `true`. If `Observable<boolean>` last
* emitted `false` then any `T`s up until the next (`Observable<T>`) `true` will be discarded. In other words, toggler
* `Observable<boolean>` controls whether or not output `Observable<T>` emits values or not. If source
* `Observable<T>` {@link complete()}s then toggler `Observable<boolean>` will be {@link unsubscribe()}d, however
* {@link complete()} from toggler `Observable<boolean>` is ignored. Calling {@link unsubscribe()} on output
* `Observable<T>` will call {@link unsubscribe()} on both source and toggler.
*
* @example <caption>Marble diagram</caption>
*
* // source s: |--1--2--3--4--5--6--7--8--|
* // source t: |-----T-----F-----T--F-----|
* // toggle(s, t): |-----2--3--4-----6--7-----|
*
* @param {ObservableT<T>} source observable that acts as source emits
* @param {ObservableT<boolean>} toggler observable that controls the emits of the returned observable
* @returns {ObservableT<T>} the observable that only emits when toggler's last emit is `true`
*/
function toggle(source, toggler) {
return new Observable(({ complete, error, next }) => {
let doNext = false;
const togglerSubscription = toggler.subscribe({
next: (bool) => (doNext = bool)
});
const sourceSubscription = source.subscribe({
complete,
error,
next: (value) => doNext && next(value)
});
return () => {
togglerSubscription.unsubscribe();
sourceSubscription.unsubscribe();
};
});
}
export { Observable, combine, combineObject, debounce, filter, flatMap, fromCallback, fromEvent, fromPromise, map, merge, of, skip, skipRepeats, startWith, switchLatest, toggle }; |
Contributor
Author
|
That's a good SO link, thanks! I've also been doing some reading into FRP and going through the react exercise, which has been helpful. I'm going to try to take a crack at using your observable code in the exercise. I'll let you know if I have any questions! |
seeksort
force-pushed
the
exercise-hangman
branch
from
November 17, 2019 14:37
b42db99 to
9c89d10
Compare
seeksort
force-pushed
the
exercise-hangman
branch
from
December 28, 2019 17:54
9c89d10 to
0b66497
Compare
Contributor
Author
|
@SleeplessByte Unfortunately, I've had trouble finding enough spare time to try to work through an FRP implementation with the library you posted. I saw that you left comments, therefore I went ahead and responded to those on the implementation as it is now. Definitely feel free to modify if you have an idea for FRP, but I didn't want to leave this PR hanging indefinitely. |
SleeplessByte
force-pushed
the
exercise-hangman
branch
from
December 16, 2020 20:59
62c9c0c to
c329ce5
Compare
Member
|
I am closing this because it is stale. If someone else picks up the exercise, they can still refer to this. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Closes #773
The problem specification mentions using functional reactive programming to solve this problem, and recommends installing an FRP library. This didn't sound like a good idea given that all exercises use the same dependencies.
Additionally, the Python version does not use FRP. The only other exercises I saw that have implemented this exercise are C#, F#, and OCaml. The first two had their own version of observables in their common libraries, while OCaml is a functional language.
I went ahead and implemented the exercise without FRP, but I'm interested if anyone has any ideas on implementing FRP without a library. I know that observables are "streams" that emit messages when activities occur, but that's the extent of my FRP knowledge.