Asynchronous Stream Publisher - Example

[ttulka]

Based on Supplier, for the case you don't know the count of elements
(for example streaming from a file, etc).

Publisher is controlled by the null value - enf of stream.

[akarnokd]

cancelled being true here is impossible because the only way to get that true is by submitting a Subscription to the Subscriber that happens once and under this condition.

[akarnokd]

The problem with crashing onXXX is that you may end up in an undefined state where even calling onError fails.

[viktorklang]

Hi @ttulka!

For my understanding, does this solve a use-case that the Iterable-based example doesn't?

[ttulka]

Hi @viktorklang
yes, this example is for the case when a "hasNext()" method is not available, it means, the stream tells us when it is over by sending a 'null' value.
A use-case could be reading lines from a file or a socket - for example using 'BufferedReader.readLine()' as the supplier.

[viktorklang]

@ttulka Ah, ok, couldn't that be solved by doing a read-ahead of 1 element? (So when the iterator() method on Iterable is called we do a read-ahead and then we advance based on that?

def supplier2Iterable[T](s: Supplier[T]): Iterable[T] = new Iterable[T] {
  override def iterator(): Iterator[T] = new Iterator[T] {
    private[this] var elem = s.get()
    override def hasNext(): Boolean = elem != null
    override def next(): T =
      if (elem == null) Iterator.empty.next()
      else {
        val prev = elem
        elem = s.get()
        prev
      }
  }

}

[ttulka]

@viktorklang This makes definitely sense, actually it is the same logic I put into the iterable publisher.
I removed my original class and created a new one - a simple extension of the iterable publisher. This is exactly the example to explain this use case.
Thank you!