title: `distinctUntilChanged` how to allow the same item once onNext is called
body: <p>I have an observable which continuously emits items ,and I need to process each one (the process function takes some time). So, meantime while processing an Item, if another item emits with the same value,
  I can ignore it, since the same is already in progress. But once the current item is processed (and called onNext). and later if the same request comes, I should allow it.
  I used the <code>distinctUntildChanged</code> operator, but what I can see is , it will not allow if the current item is same as the last one, even if the last item completed processing and called <code>onNext</code>.</p>
  
  <p>I have a sample to demonstrate the issue</p>
  
  <p>I have a class</p>
  
  <pre><code>class User {
      String id;
      String name;
  
      public User(String id, String name) {
          this.id = id;
          this.name = name;
      }
  
      @Override
      public boolean equals(Object obj) {
          User obj1 = (User) obj;
          return id.equals(obj1.id);
      }
  
      @Override
      public String toString() {
          return name;
      }
  }
  </code></pre>
  
  <p>And an observable (Subject)</p>
  
  <pre><code>Subject&lt;User&gt; mSubject = PublishSubject.create();
  </code></pre>
  
  <p>And my Observable chain is</p>
  
  <pre><code> mSubject
  .doOnNext(i -&gt; Log.d(TAG, "emitted: " + i))
  .observeOn(Schedulers.io())
  .distinctUntilChanged()
  .map(user -&gt; {
      Log.d(TAG, "processing " + user);
      Thread.sleep(5000); // processing takes 5 seconds
      return user;
  }).subscribe(user -&gt; Log.d(TAG, "onNext: " + user.name));    
  </code></pre>
  
  <p>And I emit values like this</p>
  
  <pre><code>    for (int i = 0; i &lt; 20; i++) {
          Thread.sleep(1000);            
          mSubject.onNext(new User(String.valueOf(1), "User " + i)); // all `User`s have same id
      }
  </code></pre>
  
  <p>The result is</p>
  
  <pre><code>emitted: User 0
  processing User 0
  emitted: User 1
  emitted: User 2
  emitted: User 3
  emitted: User 4
  onNext: User 0
  emitted: User 5
  emitted: User 6
  emitted: User 7
  emitted: User 8
  emitted: User 9
  emitted: User 10
  emitted: User 11
  emitted: User 12
  emitted: User 13
  emitted: User 14
  emitted: User 15
  emitted: User 16
  emitted: User 17
  emitted: User 18
  emitted: User 19
  </code></pre>
  
  <p>All the objects are same here (equals method checks id).  As you can see it took <code>user0</code> first time, and will take 5 seconds to process, during this time I can ignore incoming items,
  but after that <code>onNext: User 0</code>  I should allow the same user request, but <code>distinctUntilChanged</code> does not allow since its hoding the last value eqal to the same user, How can I do this? 
  Hope my question is clear.</p>