title: Window flowable based on a property
body: <p>I have a stream of events that contain a time (<code>Instant</code>). I want to create chunks starting from the first time of a given size (<code>Duration</code>). I also want to emit the empty chunks if nothing came in during a set time.</p>
  
  <p><strong>An example:</strong></p>
  
  <p>Input: <code>[t0, t1, t2, t3, t7, t8, t9, t10]</code> </p>
  
  <p>Output with window size 2: <code>[[t0, t1], [t2, t3], [], [t7], [t8, t9], [t10]]</code></p>
  
  <p><strong>Problem:</strong></p>
  
  <p>I have tried creating an auxiliary window publisher that will emit the instants at which the new window should be created. I am sharing what I have below for reference, but it's not functional.</p>
  
  <p><strong>Code:</strong></p>
  
  <pre><code>/**
   * Creates a [Flowable] of chunks of time. It may return empty chunks
   * if nothing happened during the set period.
   *
   * ie: [1, 2, 3, 4] -&gt; [[1, 2], [3, 4]]
   */
  fun &lt;T : TimeAware&gt; Flowable&lt;T&gt;.timeChunks(size: Duration, startAt: Instant? = null): TimeChunksFlowable&lt;T&gt; {
      return TimeChunksFlowable(this, size, startAt)
  }
  
  class TimeChunksFlowable&lt;T : TimeAware&gt; (
      private val source: Flowable&lt;T&gt;,
      val size: Duration,
      private var startAt: Instant? = null
  ) : Flowable&lt;TimeBracketFlowable&lt;T&gt;&gt;() {
  
      init {
          require(!size.isNegative &amp;&amp; !size.isZero) { "Size must be positive: $size" }
      }
  
      override fun subscribeActual(s: Subscriber&lt;in TimeBracketFlowable&lt;T&gt;&gt;) {
          source
              .publish { it.window() }
              .subscribe(s)
      }
  
      private fun Flowable&lt;T&gt;.window(): Flowable&lt;TimeBracketFlowable&lt;T&gt;&gt; {
          val chunks = chunks()
          return window(chunks).brackets(chunks)
      }
  
      private fun Flowable&lt;Flowable&lt;T&gt;&gt;.brackets(starts: Flowable&lt;Instant&gt;): Flowable&lt;TimeBracketFlowable&lt;T&gt;&gt; {
          val bi = BiFunction { f: Flowable&lt;T&gt;, i: Instant -&gt; f.timeBracket(i, i + size) }
          return zipWith(starts, bi)
      }
  
      /**
       * Creates a helper [Flowable] which emits an event every time we need to create a new chunk.
       */
      private fun Flowable&lt;T&gt;.chunks(): ChunkStartsFlowable&lt;T&gt; {
          return ChunkStartsFlowable(this, size, startAt)
      }
  
      private class ChunkStartsFlowable&lt;T : TimeAware&gt; (
          private val source: Flowable&lt;T&gt;,
          private val size: Duration,
          private val startAt: Instant?
      ) : Flowable&lt;Instant&gt;() {
  
          override fun subscribeActual(s: Subscriber&lt;in Instant&gt;) {
  
              val subscriber = InternalSubscriber&lt;T&gt;(size, startAt)
              source.subscribe(subscriber)
              subscriber.publisher.subscribe(s)
          }
  
          private class InternalSubscriber&lt;T : TimeAware&gt; (
              private val size: Duration,
              private var startAt: Instant?
          ) : Subscriber&lt;T&gt; {
  
              val publisher = UnicastProcessor.create&lt;Instant&gt;()
  
              override fun onSubscribe(s: Subscription) {
                  publisher.onSubscribe(s)
              }
  
              override fun onNext(t: T) {
                  var start = startAt ?: t.time
  
                  // add empty chunks since nothing happened
                  while (start + size &lt;= t.time) {
                      publisher.onNext(start)
                      start += size
                  }
  
                  startAt = start
              }
  
              override fun onError(t: Throwable) {
                  publisher.onError(t)
              }
  
              override fun onComplete() {
                  startAt?.apply(publisher::onNext)
                  publisher.onComplete()
              }
  
          }
  
      }
  }
  </code></pre>