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.