title: lock an operator chain while an observable item is passing through it
body: <p>I have an Observable source and an operator chain that transforms the source into a target type. Generally for each source item, up to one target is produced.</p>
  
  <pre><code>Source -&gt; Operator chain -&gt; Target
  </code></pre>
  
  <p>Operator logic is kind of complex and involves more than one async database calls using IO scheduler. I omit the details here as it does not seem to be relevant.
  What I see is that new Observables are coming from the Source while a previous observable is still being processed by the chain. So it resembles some sort of pipeline. This is probably a good thing in many cases but not in mine.</p>
  
  <p>So I am looking a way on how to delay the source items to coming into the chain (effectively locking it) until a previous item reaches the Target. Is there any known pattern of doing this?</p>
  
  <p>One ugly solution I see is to use something like this at the beginning of the chain:</p>
  
  <pre><code>zip(source, signal, (source, signal)-&gt;source)
  </code></pre>
  
  <p>where signal is a custom observable for pushing a notification into every time the chain is ready to accept a new source item (one notification initially and when an item being processed reaches the end of the  chain)
  But I find it a bit hacky. Can it be achieved more gracefully or using set of standard operators?</p>
  
  <p>Here is the synthetic example to reproduce the behavior I do not want. 
  Source is 100ms interval timer.
  Operator chain is a slow (10x slower than source) async call that computes a square on Schedulers.io()
  Target item is effectively a source squared.</p>
  
  <pre><code>Subscription s = Observable.timer(100, 100, TimeUnit.MILLISECONDS)
      .doOnNext(source-&gt;System.out.println("source: " + source))
      .concatMap(source-&gt;Observable.create(subscr-&gt;{
        Schedulers.io().createWorker().schedule(()-&gt;{
          subscr.onNext(source * source);
          subscr.onCompleted();
        }, 1000, TimeUnit.MILLISECONDS);
      }))
      .doOnNext(target-&gt;System.out.println("target: " + target))
      .subscribe();
  Thread.sleep(10000);
  s.unsubscribe();
  </code></pre>
  
  <p>Both source and target are printed out:</p>
  
  <pre><code>source: 0
  source: 1
  source: 2
  source: 3
  source: 4
  source: 5
  source: 6
  source: 7
  source: 8
  source: 9
  source: 10
  source: 11
  target: 0
  source: 12
  source: 13
  source: 14
  source: 15
  source: 16
  source: 17
  source: 18
  source: 19
  source: 20
  target: 1
  source: 21
  source: 22
  source: 23
  source: 24
  source: 25
  source: 26
  source: 27
  source: 28
  source: 29
  source: 30
  source: 31
  target: 4
  source: 32
  source: 33
  </code></pre>
  
  <p>But what I would like to achieve is:</p>
  
  <pre><code>source: 0
  target: 0
  source: 1
  target: 1
  source: 2
  target: 4
  ...
  </code></pre>