Reactive Slick

Here's the ongoing work on Reactive Slick.

/cc @cvogt for review later
On Fri, Dec 12, 2014 at 10:31 PM, Stefan Zeiger <notifications@github.com>
wrote:

> In src/main/scala/scala/slick/backend/DatabaseComponent.scala
> <https://github.com/slick/slick/pull/1021#discussion-diff-21774716>:
>
> > +      if(s ne null) {
> > +        streamState = null
> > +        if(streamLogger.isDebugEnabled) streamLogger.debug("Scheduling stream continuation after transition from demand = 0")
> > +        val a = streamingAction
> > +        database.scheduleSynchronousStreaming(a, this)(s.asInstanceOf[a.StreamState])
> > +      } else {
> > +        if(streamLogger.isDebugEnabled) streamLogger.debug("Saw transition from demand = 0, but no stream continuation available")
> > +      }
> > +    }
> > +
> > +    ////////////////////////////////////////////////////////////////////////// Subscription methods
> > +
> > +    def request(l: Long): Unit = if(!cancelRequested) {
> > +      if(l <= 0)
> > +        throw new IllegalArgumentException("Requested count must not be <= 0 (see Reactive Streams spec, 3.9)")
> > +      else if(requested + l < 0) {
>
> @viktorklang <https://github.com/viktorklang> This turns out to be
> surprisingly difficult.
>
> Actually, it seems the spec is not precise enough on this in the first
> place. Imagine the following corner case: In onSubscribe you
> request(Long.MaxValue) and in onNext you request(1). Is this allowed or
> not? In other words, is the element considered delivered *before* or
> *after* onNext?
>
What does the TCK say?
I'd say that since onX are `signals` that may as well be message sends
across a network, you need to deduct the demand -just before- you call
onNext. I think you have a point in that this should be spelled out in the
spec.


> Let's assume that that it's *before*. To comply with 3.17 we can't rely
> on the demand var, which is only updated in batches, so we have to count
> delivered elements individually.
>
Updating in batches is fine as long as you don't process `request` signals
until after you have updated the batch, or am I missing something?


> This is all fine when calling request(1) synchronously from within onNext
> but how do you make this work in the case where request(1) is called
> asynchronously? The number of pending elements would have to be an
> AtomicLong, thus requiring a memory barrier for every single element that
> is emitted.
>
You need a memory barrier in any case for async submissions otherwise you
don't have visibility.

So how about we assume that it's considered delivered only *after* onNext
> returns? The subscriber would not have any guarantee that this has actually
> happened until it gets the next signal, so it couldn't safely request(1)
> before this happens. But it could still do this asynchronously, so we're
> still out of luck.
>
> Am I overlooking some other way to make this work efficiently?
>
Do you have a benchmark showing what the impact is?


> Otherwise I think it would make sense to change the spec here, e.g. to
> allow the Publisher to limit the *total* number of requested elements
> instead of the number of peding elements. The corner case does not seem to
> have any relevance in the real world (either you request a small buffer
> window all the time, or you request Long.MaxValue once and never request
> more).
>
That means that you now require everyone to keep 2 longs (sent + requested),
however at this point, 1 week before 1.0.0.RC1, I think it's too late for
this change,
unless it's about fixing spec, interface or TCK -bugs-.
(We've been working on this over a year now)

—
> Reply to this email directly or view it on GitHub
> <https://github.com/slick/slick/pull/1021/files#r21774716>.
>



-- 
Cheers,
√