title: Flux.interval does not emits the values for new events/ object created
body: <p><code>Once the new Object is been added in a pipeline i can see the value(obj.url) of the new Object being printed without getting executing the Flux part.</code></p>
  <ul>
  <li>Note: If i restart my application, it able to read all the objects and executes completely.</li>
  <li>Issue: Seems like flux.interval is not getting subscribed for the
  new events.</li>
  </ul>
  <p>The new real time value will be added into a Map&lt;String, BaseSubscriber&gt;
  Once added it will invoke startPipeline() method where it get printed with a new value but it wont execute after that . So here the real time added objects wont be executed with <code>getMessages()</code> and <code>doSomeTask()</code> method.</p>
  <pre><code>class pipelineProcess extends BaseSubscriber&lt;Data&gt;{
  
   private final XYZ obj;
  
   @PostConstruct
   public void startPipeline() {
  
   log.info(&quot;New Objected read {} &quot;,obj.getUrl())
  
   Flux.interval(Duration.ofMillis(500))
  
                  .flatMapIterable(aLong -&gt; getMessages())
  
                  .flatMap(message -&gt; doSomeTask(message))
  
                  .subscribe(this);
  
      }
  }
   @Override
      protected void hookOnSubscribe(Subscription subscription) {
          log.info(&quot;Hook on Subscribe called for {} : &quot;,obj.getUrl());
          request(1);
      }
  
      @Override
      protected void hookOnNext(ScanRequestData next) {
          log.info(&quot;Hook on Next called for {}: &quot;, obj.getUrl());
          request(1);
      }
  
      @Override
      protected void hookOnError(Throwable throwable) {
          log.error(&quot;hookOnError exception&quot;, throwable);
          request(1);
      }
  
      @Override
      protected void hookOnComplete() {
          log.warn(&quot;The reactor pipeline of {} has been completed.&quot;, obj.getUrl());
      }
  
      @Override
      protected void hookOnCancel() {
          log.warn(&quot;The reactor pipeline of {} has been cancelled.&quot;, obj.getUrl());
      }
  }
  
  </code></pre>
  <p>Can anyone tell me how to overcome this issue</p>