RemoteChunkingManagerStepBuilder extends FaultTolerantStepBuilder<I, O> { public TaskletStep build() { Assert.notNull(this.inputChannel, "An InputChannel must be provided"); Assert.state(this.outputChannel == null || this.messagingTemplate == null, "You must specify either an outputChannel or a messagingTemplate but not both."); if (this.messagingTemplate == null) { this.messagingTemplate = new MessagingTemplate(); this.messagingTemplate.setDefaultChannel(this.outputChannel); if (this.logger.isDebugEnabled()) { this.logger.debug("No messagingTemplate was provided, using a default one"); } } ChunkMessageChannelItemWriter<O> chunkMessageChannelItemWriter = new ChunkMessageChannelItemWriter<>(); chunkMessageChannelItemWriter.setMessagingOperations(this.messagingTemplate); chunkMessageChannelItemWriter.setMaxWaitTimeouts(this.maxWaitTimeouts); chunkMessageChannelItemWriter.setThrottleLimit(this.throttleLimit); chunkMessageChannelItemWriter.setReplyChannel(this.inputChannel); super.writer(chunkMessageChannelItemWriter); return super.build(); } RemoteChunkingManagerStepBuilder(String stepName); RemoteChunkingManagerStepBuilder<I, O> inputChannel(PollableChannel inputChannel); RemoteChunkingManagerStepBuilder<I, O> outputChannel(MessageChannel outputChannel); RemoteChunkingManagerStepBuilder<I, O> messagingTemplate(MessagingTemplate messagingTemplate); RemoteChunkingManagerStepBuilder<I, O> maxWaitTimeouts(int maxWaitTimeouts); RemoteChunkingManagerStepBuilder<I, O> throttleLimit(long throttleLimit); TaskletStep build(); @Override RemoteChunkingManagerStepBuilder<I, O> reader(ItemReader<? extends I> reader); @Override RemoteChunkingManagerStepBuilder<I, O> repository(JobRepository jobRepository); @Override RemoteChunkingManagerStepBuilder<I, O> transactionManager(PlatformTransactionManager transactionManager); @Override RemoteChunkingManagerStepBuilder<I, O> listener(Object listener); @Override RemoteChunkingManagerStepBuilder<I, O> listener(SkipListener<? super I, ? super O> listener); @Override RemoteChunkingManagerStepBuilder<I, O> listener(ChunkListener listener); @Override RemoteChunkingManagerStepBuilder<I, O> transactionAttribute(TransactionAttribute transactionAttribute); @Override RemoteChunkingManagerStepBuilder<I, O> listener(org.springframework.retry.RetryListener listener); @Override RemoteChunkingManagerStepBuilder<I, O> keyGenerator(KeyGenerator keyGenerator); @Override RemoteChunkingManagerStepBuilder<I, O> retryLimit(int retryLimit); @Override RemoteChunkingManagerStepBuilder<I, O> retryPolicy(RetryPolicy retryPolicy); @Override RemoteChunkingManagerStepBuilder<I, O> backOffPolicy(BackOffPolicy backOffPolicy); @Override RemoteChunkingManagerStepBuilder<I, O> retryContextCache(RetryContextCache retryContextCache); @Override RemoteChunkingManagerStepBuilder<I, O> skipLimit(int skipLimit); @Override RemoteChunkingManagerStepBuilder<I, O> noSkip(Class<? extends Throwable> type); @Override RemoteChunkingManagerStepBuilder<I, O> skip(Class<? extends Throwable> type); @Override RemoteChunkingManagerStepBuilder<I, O> skipPolicy(SkipPolicy skipPolicy); @Override RemoteChunkingManagerStepBuilder<I, O> noRollback(Class<? extends Throwable> type); @Override RemoteChunkingManagerStepBuilder<I, O> noRetry(Class<? extends Throwable> type); @Override RemoteChunkingManagerStepBuilder<I, O> retry(Class<? extends Throwable> type); @Override RemoteChunkingManagerStepBuilder<I, O> stream(ItemStream stream); @Override RemoteChunkingManagerStepBuilder<I, O> chunk(int chunkSize); @Override RemoteChunkingManagerStepBuilder<I, O> chunk(CompletionPolicy completionPolicy); @Override RemoteChunkingManagerStepBuilder<I, O> writer(ItemWriter<? super O> writer); @Override RemoteChunkingManagerStepBuilder<I, O> readerIsTransactionalQueue(); @Override RemoteChunkingManagerStepBuilder<I, O> listener(ItemReadListener<? super I> listener); @Override RemoteChunkingManagerStepBuilder<I, O> listener(ItemWriteListener<? super O> listener); @Override RemoteChunkingManagerStepBuilder<I, O> chunkOperations(RepeatOperations repeatTemplate); @Override RemoteChunkingManagerStepBuilder<I, O> exceptionHandler(ExceptionHandler exceptionHandler); @Override RemoteChunkingManagerStepBuilder<I, O> stepOperations(RepeatOperations repeatTemplate); @Override RemoteChunkingManagerStepBuilder<I, O> startLimit(int startLimit); @Override RemoteChunkingManagerStepBuilder<I, O> listener(StepExecutionListener listener); @Override RemoteChunkingManagerStepBuilder<I, O> allowStartIfComplete(boolean allowStartIfComplete); @Override RemoteChunkingManagerStepBuilder<I, O> processor(ItemProcessor<? super I, ? extends O> itemProcessor);  }
@Test public void testMandatoryInputChannel() { RemoteChunkingManagerStepBuilder<String, String> builder = new RemoteChunkingManagerStepBuilder<>("step"); this.expectedException.expect(IllegalArgumentException.class); this.expectedException.expectMessage("An InputChannel must be provided"); TaskletStep step = builder.build(); }