title: Project Reactor: cache last item for each subscribed publisher
body: <p>I have a processor which subscribes to publishers which arrive in arbitrary time. For each new subscriber to the processor, I want to emit the last item from each publisher.</p>
  
  <pre><code>    class PublishersState {
          val outputProcessor = DirectProcessor.create&lt;String&gt;()
  
          fun addNewPublisher(publisher: Flux&lt;String&gt;) {
              publisher.subscribe(outputProcessor)
          }
  
          fun getAllPublishersState(): Flux&lt;String&gt; = outputProcessor
      }
  
      val publisher1 = Mono
          .just("Item 1 publisher1")
          .mergeWith(Flux.never())
  
      val publisher2 = Flux
          .just("Item 1 publisher2", "Item 2 publisher2")
          .mergeWith(Flux.never())
  
      val publishersState = PublishersState()
  
      publishersState.getAllPublishersState().subscribe {
          println("Subscriber1: $it")
      }
  
      publishersState.addNewPublisher(publisher1)
  
      publishersState.addNewPublisher(publisher2)
  
      publishersState.getAllPublishersState().subscribe {
          println("Subscriber2: $it")
      }
  </code></pre>
  
  <p>I need to change the code above so it will output the following:</p>
  
  <pre><code>Subscriber1: Item 1 publisher1
  Subscriber1: Item 1 publisher2
  Subscriber1: Item 2 publisher2
  // Subscriber2 subscribers here and receives the last item from each publisher
  Subscriber2: Item 1 publisher1
  Subscriber2: Item 2 publisher2
  </code></pre>
  
  <p>Is there a simple way to cache the last item for each publisher?</p>