title: Spring Integration Java DSL flow Splitter/Aggregator delete file after processing all lines
body: <p>Using Spring Integration Java DSL, I have constructed a flow where I'm processing files synchronously with a <code>FileSplitter</code>.  I've been able to use the <code>setDeleteFiles</code> flag on a <code>AbstractFilePayloadTransformer</code> to delete the file after converting each line in <code>File</code> to a <code>Message</code> for subsequent processing, like so:</p>
  
  <pre><code>@Bean
  protected IntegrationFlow s3ChannelFlow() {
      // do not exhaust filesystem w/ files downloaded from S3
      FileToInputStreamTransformer transformer = new FileToInputStreamTransformer();
      transformer.setDeleteFiles(true);
  
      // @see http://docs.spring.io/spring-integration/reference/html/files.html#file-reading
      // @formatter:off
      return IntegrationFlows
          .from(s3Channel())
          .channel(StatsUtil.createRunStatsChannel(runStatsRepository))
          .transform(transformer)
          .split(new FileSplitter())
          .transform(new JsonToObjectViaTypeHeaderTransformer(new Jackson2JsonObjectMapper(objectMapper), typeSupport))
          .publishSubscribeChannel(p -&gt; p.subscribe(persistenceSubFlow()))
          .get();
      // @formatter:on
  }
  </code></pre>
  
  <p>This works fine, but is slow.  So I attempt to add an <code>ExecutorChannel</code> after the <code>.split</code> above, like so:</p>
  
  <pre><code>.channel(c -&gt; c.executor(Executors.newFixedThreadPool(10)))
  </code></pre>
  
  <p>But then the aforementioned delete flag does not allow the flow to complete successfully deleting file(s) before they are completely read.</p>
  
  <p>If I remove the flag I have the potential to exhaust the local file system where files were synchronized from S3.</p>
  
  <p><strong>What could I introduce above to a) process each file completely and b) delete file from local filesystem once done?</strong>  In other words, is there a way to get to know exactly when a file is completely processed (when it's lines have been processed asynchronously via threads in a pool)?</p>
  
  <p>If you're curious here's my impl of <code>FileToInputStreamTransformer</code>:</p>
  
  <pre><code>public class FileToInputStreamTransformer extends AbstractFilePayloadTransformer&lt;InputStream&gt; {
  
      private static final int BUFFER_SIZE = 64 * 1024; // 64 kB
  
      @Override
      // @see http://java-performance.info/java-io-bufferedinputstream-and-java-util-zip-gzipinputstream/
      protected InputStream transformFile(File payload) throws Exception {
          return new GZIPInputStream(new FileInputStream(payload), BUFFER_SIZE);
      }
  }
  </code></pre>
  
  <p><strong>UPDATE</strong></p>
  
  <p>So how does something in downstream flow know what to ask for? </p>
  
  <p>Incidentally, if I'm following your advice correctly, when I update the <code>.split</code> with <code>new FileSplitter(true, true)</code> above, I get </p>
  
  <pre>
  2015-10-20 14:26:45,288 [pool-6-thread-1] org.springframework.integration.handler.LoggingHandler ERROR org.springframework.integration.transformer.MessageTransformationException: failed to transform message; nested exception is java.lang.IllegalArgumentException: 'json' argument must be an instance of: [class java.lang.String, class [B, class java.io.File, class java.net.URL, class java.io.InputStream, class java.io.Reader] , but gotten: class org.springframework.integration.file.splitter.FileSplitter$FileMarker
      at org.springframework.integration.transformer.AbstractTransformer.transform(AbstractTransformer.java:44)
  </pre>