As a general rule, I abstained from adding names in source files - you can add yourself to the AUTHORS file.

Sorry about this, just IntelliJ template, fixing it

I'm obsessed with indentation and I'd prefer for this to be consistent with the other files. Personally I don't like things dangling so far on the right, although I can understand the preference.

Doing a flatMap on a Future can be pretty inefficient since it forces an asynchronous boundary.

This is why Monix introduces a couple of special Future extension methods for avoiding async boundaries if the Future returned by onNext is already complete. Monix is efficient because it tries to collapse these async boundaries if the downstream subscriber returns a plain Continue or Stop that's immediately available.

Don't get me wrong, this code is correct, but it can be more efficient if you use syncFlatMap instead:

out.onNext(separator).syncFlatMap {
  ???
}

As a note, for user code this is actually dangerous because it is stack-unsafe (in loops), but we know what we are doing here and it's fine.

Just as with firstTime, you're only modifying isDone in onComplete and onError. The calls to onNext, onComplete and onError are ordered (synchronized) already so this can be a private[this] var.

The only concurrency problem you can have is that the producer doesn't need to back-pressure for that final Future[Ack] returned by the last onNext, so depending what you do in those flatMaps, you can end up with a concurrency problem, but you don't need anything special here.

In this case you don't need to back-pressure on downstreamAck, because this is legal (as an optimization of the protocol:

out.onNext(e)
out.onComplete()

But this is NOT legal of course:

out.onNext(e)
out.onNext(e)
out.onComplete()

There are operators that keep that last ack around and use it in onComplete, but that's only because they have to do more stuff.

I suspect that if you have to stream a final end, then you do need to have that ack to back-pressure on, but you don't need that here.

If we introduce this, we also need the version with the start and the end tokens and I think it can be handled by the same implementation - in that Observable source you can make them optional or something.

👍 it's good that you introduced the test as well

You are modifying firstTime in onNext only, so you don't need to synchronize it, because the onNext calls are guaranteed to be ordered (the same guarantee you get with actors processing messages).

Atomics come with overhead and this can be a simple private[this] var in that Subscriber class.

I didn't know about this, good to know

Looking good dude. One final touch ... there's no need for back-pressure with this syncOnContinue here, because you can do ...

out.onNext(end.get)
out.onComplete()

This optimisation (not waiting on the final onNext for sending onComplete) is good from a networking POV, but it's also why Observable.tailRecM can work well.

I would also prefer if you wouldn't wait on downstreamAck in case end.isDefined == false. It's totally fine if we have multiple if branches or case statements.

Ok then :-). I'm still learning about all these things. You have been a great help

It's totally fine @omainegra, no rush, I breathe this protocol and know the optimisation opportunities because I came up with it.

And your first try was correct and pretty good.

I need assistance here please. If I don't do back-pressure before onComplete, one unit test fails.

[info] monix.reactive.internal.operators.IntersperseSuite
[info] - should emit exactly the requested elements
[info] - should work for synchronous observers
[info] - should work for asynchronous observers *** FAILED ***
[info]   received 24310 != expected 24530 (BaseOperatorSuite.scala:157)
[info]     minitest.api.Asserts$class.assertEquals(Asserts.scala:67)
[info]     monix.reactive.internal.operators.BaseOperatorSuite.assertEquals(BaseOperatorSuite.scala:30)
[info]     monix.reactive.internal.operators.BaseOperatorSuite$$anonfun$3.apply(BaseOperatorSuite.scala:157)
[info]     monix.reactive.internal.operators.BaseOperatorSuite$$anonfun$3.apply(BaseOperatorSuite.scala:125)
[info]     minitest.api.TestSpec$$anonfun$from$1.apply(TestSpec.scala:54)
[info]     minitest.api.TestSpec$$anonfun$from$1.apply(TestSpec.scala:52)
[info]     minitest.api.TestSpec.apply(TestSpec.scala:28)

Sure, but please copy/paste your newest onComplete

It does look like a contract problem, I don't dismiss the possibility of me being wrong :-)

This is the new onComplete. In the else branch is commented the code with back-pressure. As it is, it's not working.

def onComplete() = {
  if (end.isDefined){
    downstreamAck.syncOnComplete {
      case Success(_) =>
        out.onNext(end.get)
        signalOnComplete()
      case _ =>
    }
  } else {
    signalOnComplete()
//          downstreamAck.syncOnComplete {
//            case Success(_) => signalOnComplete()
//            case _ =>
//          }
  }
}

private[this] def signalOnComplete(): Unit = {
  if (!isDone) {
    isDone = true
    out.onComplete()
  }
}

OK, I think I know what's going on. In onNext you can end up pushing 2 events instead of one, so if the subscriber is delaying on sep, you still have the elem queued up for sending it, so you can end up triggering onNext(end) and onComplete before that happens. See, I was wrong :-)

Btw, you need to check that you receive a Continue and don't need to do anything in case of Stop or Failure(ex), so doing just one syncOnContinue there is totally fine.

ack.syncOnContinue {
  if (end.nonEmpty) out.onNext(end.get)
  out.onComplete
}

Or smth.

Yes, It's as you say. One last question. Is it required the isDone logic`?, because I just removed it and all test passed OK

Yeah, you probably don't need it - isDone in onComplete is usually useful when you're doing stuff in onNext that triggers a completion and then you want to ensure that you're not going to be completing that stream a second time.

It might be OK to remove it.

👍 looking good

👍