My thoughts about single events in stream

[romansl]

I agree with author of issue #39, but found some disadvantages.

I dont like the idea of loosing events in StreamSink and merge. The events contains important internal data. You can not just throw them away. For example this logic will be broken:

val sink = streamSink<Command>()
val state= cellLoop<State>()

val anotherCommands = sink.snapshot(state).filter {
   // need to produce another command
}

state loop sink.merge(anotherCommands).accum(...)

Also I think that the transaction complements things like merge and you no longer need transaction without multiple events in stream.

This example becomes error:

val sink = streamSink<Command>()

Sodium.tx {
   sink.send(BeginCommand())
   sink.send(SomeCommand())
   sink.send(EndCommand())
}

You can rewrite it using streamSink<List<Command>>, but Sodum.tx now becomes useless. Even more, now it is dangerous.

[pyrtsa]

My recommendation is to remove coalesce and (the combiner-less overload of) merge altogether¹, and to make StreamSink assert when more than one event is being sent with sink.send(value).

@romansl I think you've got a point we can solve there, but I'm not entirely following – what exactly makes Sodium.tx dangerous here?

¹) For convenience, the library could provide the left and right biased combiner functions separately.

[romansl]

The marge without arguments can combine events to Pair<A, A> or List<A>(2).

Also marge can have signature Stream<A>.marge(b: Stream<B>, combiner: (A, B) -> C).

Transaction is dangerous because can accidentally loose important events.

[pyrtsa]

I don't think Pair<A, A> is a good default. You'd only merge two things like that. But then when you have 3 streams you want to merge, you'd end up with something like Pair<A, Pair<A, A>>. …Let alone N streams to merge!

The merging of streams has always been closely related to monoids or actually, since there never ever is a stream firing with nothing – that'd be absurd! – semigroups.

With that thought in mind, the way to merge things into a List<A> would be to map each of the inputs into single-element List<A>'s, and then merge those with list concatenation as the combiner.

In Haskell, we could even redefine the old merge conditionally as:

merge :: Semigroup a => Stream a -> Stream a -> Stream a
merge x y = mergeWith (<>) x y  -- or, for short: merge = mergeWith (<>)

which is exactly what I described above.

Further, Stream a could be made conditionally a Monoid if there's a default way to append a's together:

instance (Semigroup a) => Monoid (Stream a) where
    mempty = never
    mappend = merge

[pyrtsa]

Transaction is dangerous because can accidentally loose important events.

Yeah, I'd try hard to make it impossible to send to the same stream more than once per transaction. If it can't be done with the interface, asserting a precondition would be my second best suggestion:

val sink = streamSink<Command>()

Sodium.tx {
    sink.send(BeginCommand())
    sink.send(SomeCommand()) // throws Exception: "StreamSink already sent"
    // ...
}

[romansl]

Throwing exceptions makes it even worse. That if this happens inside logic construction?

And here we got that firing of the event in one part of code may affect another.

The send must be as safe as possible.

[romansl]

May be send must produce new transaction like defer?

[pyrtsa]

I think we come from different schools of thought here, and there's no Proven Right Way™ to write real-world FRP applications yet, so I value your ideas. Would like to understand them better.

I was talking about enforcing a precondition against sending multiple firings into a StreamSink. Whether it's an Exception or just Log.fatal("...") is irrelevant.

And here we got that firing of the event in one part of code may affect another.

In my view, transactions should be very local beasts that do as little as possible, stuff like update the mouse state into what it currently is or such. One transaction shouldn't contain many "parts of code" (but I'd be interested in seeing what the counterexample would be).

That if this happens inside logic construction?

What do you mean by inside logic construction?

[the-real-blackh]

I've checked in ed0c4037fd70dac8b8727f48a2f8b76d51d5c6d2 which changes it so an exception is now thrown if send() is called more than once per transaction. Now we have this:

• coalesce() doesn't exist in the API.
• send() throws an exception by default if called more than once per transaction. This will generally not be a problem during a "one big transaction" program initialization because the code will tend to be deterministic so the bug will tend to be caught during testing. It will never be a problem where a transaction is started implicitly.
• Passing a combining function to StreamSink() allows you to change this policy.
• s1.merge(s2) without f will throw events away. This may seem dangerous but it's a tradeoff against convenience, because the case where a merge is done when the user knows the event can never be simultaneous is extremely common. Use of the Unit type here (where throwing one away doesn't matter) is also common. Convenience of the user interface is important to me because I see Sodium as competing for mainstream approval against the broken FRP implementations (Rx and friends).

Other options I didn't choose:

1. s1.merge(s2) throws an exception. I think in a practical situation this is more dangerous than dropping the s1 event.
2. static <A extends SemiGroup> Stream.merge(Stream<A> s1, Stream<A> s2) would be extremely cumbersome in Java but the equivalent Haskell with typeclasses would work and I will most likely use that. Haskell programmers are used to much stricter APIs than Java programmers are.

So let me know if you think we can do better than this.

[ritschwumm]

how about naming s1.merge(s2) "orElse" instead? it more clearly describes what the method actually does. and of course i'm used to scala's Option#orElse ...

[the-real-blackh]

That's certainly possible, but in the non-simultaneous case it's still a merge operation. Hmm.... I'm starting to like the idea.

[romansl]

It is good idea I think.

[the-real-blackh]

merge() renamed to orElse() as suggested.