title: Move file from inbound adapter after publish subscribe flow
body: <p>I'm trying to implement the following flow:
  1) files are read from inbound adapter
  2) they are send to different flows using publish-subscribe channel with applied sequence
  3) file is moved after all the subscriber flows are ready</p>
  
  <p>This is the main flow</p>
  
  <pre><code>return IntegrationFlows
                  .from(Files.inboundAdapter(inboundOutDirectory)
                             .regexFilter(pattern)
                             .useWatchService(true)
                             .watchEvents(FileReadingMessageSource.WatchEventType.CREATE),
                          e -&gt; e.poller(Pollers.fixedDelay(period)
                                               .taskExecutor(Executors.newFixedThreadPool(poolSize))
                                               .maxMessagesPerPoll(maxMessagesPerPoll)))
                  .publishSubscribeChannel(s -&gt; s
                          .applySequence(true)
                          .subscribe(f -&gt; f
                                  .transform(Files.toStringTransformer())
                                  .&lt;String&gt;handle((p, h) -&gt; {
                                         return "something"
                                      }
                                  })                                
                                  .channel("consolidateFlow.input"))
                          .subscribe(f -&gt; f
                                  .transform(Files.toStringTransformer())
                                  .handle(Http.outboundGateway(testUri)
                                              .httpMethod(HttpMethod.GET)
                                              .uriVariable("text", "payload")                                            .expectedResponseType(String.class))
                                  .&lt;String&gt;handle((p, h) -&gt; {
                                      return "something";
                                  })
                                  .channel("consolidateFlow.input")))
                  .get();
  </code></pre>
  
  <p>And the aggregation:</p>
  
  <pre><code>public IntegrationFlow consolidateFlow()
  return flow -&gt; flow
                  .aggregate()
                  .&lt;List&lt;String&gt;&gt;handle((p, h) -&gt; "something").log()
      }
  }
  </code></pre>
  
  <p>Using the following code in the main flow after publish-subscribe</p>
  
  <pre><code>.handle(Files.outboundGateway(this.inboundProcessedDirectory).deleteSourceFiles(true))
  </code></pre>
  
  <p>ends up with</p>
  
  <p><code>Caused by: org.springframework.messaging.core.DestinationResolutionException: no output-channel or replyChannel header available</code></p>
  
  <p>If I go with this the consolidation/aggregation flow won't be reached at all.</p>
  
  <pre><code>.handle(Files.outboundAdapter(this.inboundProcessedDirectory))
  </code></pre>
  
  <p>Any idea how I could solve it? Currently I'm moving the file after the aggregation by reading the original file name from the header but it doesn't seem to be the right solution.
  I was also thinking about applying spec/advice to the inbound adapter with <em>success</em> logic to move the file but not sure whether that's the right approach.</p>
  
  <p><strong><em>EDIT1</em></strong>
  As suggested by Artem, I've added another subscriber to the publish-subscribe as follows:</p>
  
  <pre><code>...
  .channel("consolidateNlpFlow.input"))
                          .subscribe(f -&gt; f
                                  .handle(Files.outboundAdapter(this.inboundProcessedDirectory).deleteSourceFiles(true))
  ...
  </code></pre>
  
  <p>The files is moved properly, but the consolidateFlow is not being executed at all. Any idea?
  I've also tried adding the channel to the new flow <code>.channel("consolidateNlpFlow.input")</code> but it didn't change the behavior.</p>