class RemoteChunkingWorkerBuilder<I, O> {

	private static final String SERVICE_ACTIVATOR_METHOD_NAME = "handleChunk";

	private ItemProcessor<I, O> itemProcessor;

	private ItemWriter<O> itemWriter;

	private MessageChannel inputChannel;

	private MessageChannel outputChannel;

	/**
	 * Set the {@link ItemProcessor} to use to process items sent by the manager step.
	 * @param itemProcessor to use
	 * @return this builder instance for fluent chaining
	 */
	public RemoteChunkingWorkerBuilder<I, O> itemProcessor(ItemProcessor<I, O> itemProcessor) {<FILL_FUNCTION_BODY>}

	/**
	 * Set the {@link ItemWriter} to use to write items sent by the manager step.
	 * @param itemWriter to use
	 * @return this builder instance for fluent chaining
	 */
	public RemoteChunkingWorkerBuilder<I, O> itemWriter(ItemWriter<O> itemWriter) {
		Assert.notNull(itemWriter, "itemWriter must not be null");
		this.itemWriter = itemWriter;
		return this;
	}

	/**
	 * Set the input channel on which items sent by the manager are received.
	 * @param inputChannel the input channel
	 * @return this builder instance for fluent chaining
	 */
	public RemoteChunkingWorkerBuilder<I, O> inputChannel(MessageChannel inputChannel) {
		Assert.notNull(inputChannel, "inputChannel must not be null");
		this.inputChannel = inputChannel;
		return this;
	}

	/**
	 * Set the output channel on which replies will be sent to the manager step.
	 * @param outputChannel the output channel
	 * @return this builder instance for fluent chaining
	 */
	public RemoteChunkingWorkerBuilder<I, O> outputChannel(MessageChannel outputChannel) {
		Assert.notNull(outputChannel, "outputChannel must not be null");
		this.outputChannel = outputChannel;
		return this;
	}

	/**
	 * Create an {@link IntegrationFlow} with a {@link ChunkProcessorChunkHandler}
	 * configured as a service activator listening to the input channel and replying on
	 * the output channel.
	 * @return the integration flow
	 */
	@SuppressWarnings({ "unchecked", "rawtypes" })
	public IntegrationFlow build() {
		Assert.notNull(this.itemWriter, "An ItemWriter must be provided");
		Assert.notNull(this.inputChannel, "An InputChannel must be provided");
		Assert.notNull(this.outputChannel, "An OutputChannel must be provided");

		if (this.itemProcessor == null) {
			this.itemProcessor = new PassThroughItemProcessor();
		}
		SimpleChunkProcessor<I, O> chunkProcessor = new SimpleChunkProcessor<>(this.itemProcessor, this.itemWriter);

		ChunkProcessorChunkHandler<I> chunkProcessorChunkHandler = new ChunkProcessorChunkHandler<>();
		chunkProcessorChunkHandler.setChunkProcessor(chunkProcessor);

		return IntegrationFlow.from(this.inputChannel)
			.handle(chunkProcessorChunkHandler, SERVICE_ACTIVATOR_METHOD_NAME)
			.channel(this.outputChannel)
			.get();
	}

}

		Assert.notNull(itemProcessor, "itemProcessor must not be null");
		this.itemProcessor = itemProcessor;
		return this;
	
<no_super_class>