Stream Extras

[spacejack]

Description

This issue is to discuss potential additions and enhancements for mithril/stream.

This is a collection of helpers I've found useful working with Streams: https://github.com/spacejack/mithril-stream-extra It's a bit TypeScript-heavy because I wanted to add a ReadonlyStream type, but otherwise I think it's also got useful helpers for plain JS.

(Revisiting it however, I found the ReadonlyStream type is now broken with the current Typescript compiler... I'm having trouble coming up with a Stream-compatible read-only type so that ReadonlyStream types are usable with all Stream functions. This might need to be done as a complete custom stream .d.ts or brought into core to work.)

In addition to the above, these functions are for getting promises from streams:

/**
 * Promise that resolves on stream's initial value
 * Credit to @isiahmeadows for eliminating the extra map
 */
export function one<T>(s: ReadonlyStream<T>): Promise<T> {
    return new Promise<T>(resolve => {
        let done = false
        let s1: Stream<T>
        s1 = s.map(v => {
            if (done) {
                return
            }
            done = true
            if (s1 != null) {
                s1.end(true)
            }
            resolve(v)
        })
        if (done) {
            s1.end(true)
        }
    })
}

/**
 * Promise that resolves on stream's next value
 */
export function nextOne<T>(s: ReadonlyStream<T>): Promise<T> {
    return one(dropInitial(s))
}

Why

Having used streams in various ways, this is a set of helpers I've found most useful. I'd also like to hear from others about what additional stream functions or patterns they're using. And any suggestions/critiques or antipattern warnings about the above are welcome as well.

[dead-claudia]

Thought I'd give some initial feedback on each of your suggestions:

• You have a lift in your library, but it carries the same exact semantics as our Stream.lift, so we don't really need to add it.

• Your readOnly stream never terminates the returned stream, but you could fix this by just returning stream.map(x => x).

• Your dropRepeats never terminates the returned stream, but you could fix the bug while simplifying it a bit using Stream.SKIP/Stream.HALT.

• Your dropInitial never terminates the returned stream, but it could be similarly fixed and simplified using a simple boolean:

function dropInitial(s) {
    var ready = false
    return s.map(x => {
        if (ready) return x
        ready = true
        return Stream.HALT
    })
}

// This generalizes well
function dropInitial(s, n) {
    n = Math.floor(n)
    return s.map(x => {
        if (n === 0) return x
        n--
        return Stream.HALT
    })
}

• WRT your one, it could be simplified further to drop a slot:

var sentinel = {}
export function one(s) {
    return new Promise(function (resolve) {
        var child = sentinel
        var dep = s.map(function (v) {
            var memo = child
            child = undefined
            if (memo != null) {
                resolve(v)
                if (sentinel !== memo) memo.end(true)
            }
            return Stream.HALT
        })
        if (child == null) dep.end(true)
        else child = dep
    })
}

Edit: forgot to highlight a code block.

[tzkmx]

I'd like to see other operators as well, or having it easier to plug-in with an extension. For example I've followed the idea of @spacejack but borrowed inspiration from xstream/extras (same operator dropRepeats, but with an optional customEquality function)

So I came to a project with this addition: https://gist.github.com/tzkmx/d82789cabb1c635fc10f5233ace72171 ). I didn't know how to plugit into the custom library, so I've put my custom version of mithril-stream with my custom operator embedded in the source.

I'd like to see another operators added To, like groupBy, filter, etc. I know all of this can be custom implemented with map, but even then, it'd be nice to being able to publish contributions compatible with mithril-stream without forcing bloat into the goals of the project.

[dead-claudia]

@tzkmx To clarify, which do you mean by groupBy?

• A reduction like Lodash (and Java), returning a single grouped value.
• A stream emitting {key, value: Stream<T>} pairs not unlike RxJS, where each valueStream emits the values corresponding to the key it was emitted with.

The first is somewhat straightforward and simple with ES6:

function groupBy(s, func) {
    s = s.map(v => [func(v), v)])
    let acc = new Map(), result = Stream(), end = false
    s.end.map(t => {
        if (!end && t === true) {
            end = true
            result(Array.from(acc))
            result.end(true)
        }
    })
    s.map(v => {
        if (!end) {
            let values = acc.get(key)
            if (values == null) acc.set(key, values = [])
            values.push(v)
        }
        return Stream.SKIP
    })
    return result
}

The second is much less simple, and it's complicated enough I'm not going to show the code for it here.

[spacejack]

@tzkmx your dropRepeats with a callback sounds like flyd's filter. flyd is the library I usually reference for ideas.