Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

BATCH-2241 & BATCH-2243 - add namespace support for remote chunking

Browse Source
This commit is contained in:
Chris Schaefer
2014-06-12 03:14:34 -04:00
parent deb4c6fb5b
commit 4f32f623f0
18 changed files with 926 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/BatchIntegrationNamespaceHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -21,15 +21,16 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
* The namespace handler for the Spring Batch Integration namespace.
*
* @author Gunnar Hillert
* @author Chris Schaefer
* @since 1.3
*
*/
public class BatchIntegrationNamespaceHandler extends AbstractIntegrationNamespaceHandler {
/* (non-Javadoc)
* @see org.springframework.beans.factory.xml.NamespaceHandler#init()
*/
public void init() {
this.registerBeanDefinitionParser("job-launching-gateway", new JobLaunchingGatewayParser());
this.registerBeanDefinitionParser("remote-chunking-master", new RemoteChunkingMasterParser());
this.registerBeanDefinitionParser("remote-chunking-slave", new RemoteChunkingSlaveParser());
}
}

83
spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParser.java Normal file
View File

@@ -0,0 +1,83 @@
/*
* Copyright 2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.integration.config.xml;
import org.springframework.batch.integration.chunk.ChunkMessageChannelItemWriter;
import org.springframework.batch.integration.chunk.RemoteChunkHandlerFactoryBean;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.util.Assert;
import org.w3c.dom.Element;
/**
* <p>
* Parser for the remote-chunking-master namespace element.
* </p>
*
* @author Chris Schaefer
* @since 3.1
*/
public class RemoteChunkingMasterParser extends AbstractBeanDefinitionParser {
private static final String MESSAGE_TEMPLATE_ATTRIBUTE = "message-template";
private static final String STEP_ATTRIBUTE = "step";
private static final String REPLY_CHANNEL_ATTRIBUTE = "reply-channel";
private static final String MESSAGING_OPERATIONS_PROPERTY = "messagingOperations";
private static final String REPLY_CHANNEL_PROPERTY = "replyChannel";
private static final String CHUNK_WRITER_PROPERTY = "chunkWriter";
private static final String STEP_PROPERTY = "step";
private static final String CHUNK_HANDLER_BEAN_NAME_PREFIX = "remoteChunkHandlerFactoryBean_";
@Override
public AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) {
String id = element.getAttribute(ID_ATTRIBUTE);
Assert.hasText(id, "The id attribute must be specified");
String messageTemplate = element.getAttribute(MESSAGE_TEMPLATE_ATTRIBUTE);
Assert.hasText(messageTemplate, "The message-template attribute must be specified");
String step = element.getAttribute(STEP_ATTRIBUTE);
Assert.hasText(step, "The step attribute must be specified");
String replyChannel = element.getAttribute(REPLY_CHANNEL_ATTRIBUTE);
Assert.hasText(replyChannel, "The reply-channel attribute must be specified");
BeanDefinitionRegistry beanDefinitionRegistry = parserContext.getRegistry();
BeanDefinition chunkMessageChannelItemWriter =
BeanDefinitionBuilder
.genericBeanDefinition(ChunkMessageChannelItemWriter.class)
.addPropertyReference(MESSAGING_OPERATIONS_PROPERTY, messageTemplate)
.addPropertyReference(REPLY_CHANNEL_PROPERTY, replyChannel)
.getBeanDefinition();
beanDefinitionRegistry.registerBeanDefinition(id, chunkMessageChannelItemWriter);
BeanDefinition remoteChunkHandlerFactoryBean =
BeanDefinitionBuilder
.genericBeanDefinition(RemoteChunkHandlerFactoryBean.class)
.addPropertyValue(CHUNK_WRITER_PROPERTY, chunkMessageChannelItemWriter)
.addPropertyValue(STEP_PROPERTY, step)
.getBeanDefinition();
beanDefinitionRegistry.registerBeanDefinition(CHUNK_HANDLER_BEAN_NAME_PREFIX + step, remoteChunkHandlerFactoryBean);
return null;
}
}

117
spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParser.java Normal file
View File

@@ -0,0 +1,117 @@
/*
* Copyright 2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.integration.config.xml;
import org.springframework.batch.core.step.item.SimpleChunkProcessor;
import org.springframework.batch.integration.chunk.ChunkProcessorChunkHandler;
import org.springframework.batch.item.support.PassThroughItemProcessor;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.ServiceActivatorFactoryBean;
import org.springframework.integration.config.xml.AbstractConsumerEndpointParser;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* <p>
* Parser for the remote-chunking-slave namespace element. If an
* {@link org.springframework.batch.item.ItemProcessor} is not provided, an
* {@link org.springframework.batch.item.support.PassThroughItemProcessor} will be
* configured.
* </p>
*
* @author Chris Schaefer
* @since 3.1
*/
public class RemoteChunkingSlaveParser extends AbstractBeanDefinitionParser {
private static final String INPUT_CHANNEL_ATTRIBUTE = "input-channel";
private static final String OUTPUT_CHANNEL_ATTRIBUTE = "output-channel";
private static final String ITEM_PROCESSOR_ATTRIBUTE = "item-processor";
private static final String ITEM_WRITER_ATTRIBUTE = "item-writer";
private static final String ITEM_PROCESSOR_PROPERTY_NAME = "itemProcessor";
private static final String ITEM_WRITER_PROPERTY_NAME = "itemWriter";
private static final String CHUNK_PROCESSOR_PROPERTY_NAME = "chunkProcessor";
private static final String CHUNK_PROCESSOR_CHUNK_HANDLER_BEAN_NAME_PREFIX = "chunkProcessorChunkHandler_";
@Override
protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) {
String id = element.getAttribute(ID_ATTRIBUTE);
Assert.hasText(id, "The id attribute must be specified");
String inputChannel = element.getAttribute(INPUT_CHANNEL_ATTRIBUTE);
Assert.hasText(inputChannel, "The input-channel attribute must be specified");
String outputChannel = element.getAttribute(OUTPUT_CHANNEL_ATTRIBUTE);
Assert.hasText(outputChannel, "The output-channel attribute must be specified");
String itemProcessor = element.getAttribute(ITEM_PROCESSOR_ATTRIBUTE);
String itemWriter = element.getAttribute(ITEM_WRITER_ATTRIBUTE);
Assert.hasText(itemWriter, "The item-writer attribute must be specified");
BeanDefinitionRegistry beanDefinitionRegistry = parserContext.getRegistry();
BeanDefinitionBuilder chunkProcessorBuilder =
BeanDefinitionBuilder
.genericBeanDefinition(SimpleChunkProcessor.class)
.addPropertyReference(ITEM_WRITER_PROPERTY_NAME, itemWriter);
if(StringUtils.hasText(itemProcessor)) {
chunkProcessorBuilder.addPropertyReference(ITEM_PROCESSOR_PROPERTY_NAME, itemProcessor);
} else {
chunkProcessorBuilder.addPropertyValue(ITEM_PROCESSOR_PROPERTY_NAME, new PassThroughItemProcessor());
}
BeanDefinition chunkProcessorChunkHandler =
BeanDefinitionBuilder
.genericBeanDefinition(ChunkProcessorChunkHandler.class)
.addPropertyValue(CHUNK_PROCESSOR_PROPERTY_NAME, chunkProcessorBuilder.getBeanDefinition())
.getBeanDefinition();
beanDefinitionRegistry.registerBeanDefinition(CHUNK_PROCESSOR_CHUNK_HANDLER_BEAN_NAME_PREFIX + id, chunkProcessorChunkHandler);
new ServiceActivatorParser(id).parse(element, parserContext);
return null;
}
private static class ServiceActivatorParser extends AbstractConsumerEndpointParser {
private static final String TARGET_METHOD_NAME_PROPERTY_NAME = "targetMethodName";
private static final String TARGET_OBJECT_PROPERTY_NAME = "targetObject";
private static final String HANDLE_CHUNK_METHOD_NAME = "handleChunk";
private static final String CHUNK_PROCESSOR_CHUNK_HANDLER_BEAN_NAME_PREFIX = "chunkProcessorChunkHandler_";
private String id;
public ServiceActivatorParser(String id) {
this.id = id;
}
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ServiceActivatorFactoryBean.class);
builder.addPropertyValue(TARGET_METHOD_NAME_PROPERTY_NAME, HANDLE_CHUNK_METHOD_NAME);
builder.addPropertyValue(TARGET_OBJECT_PROPERTY_NAME, new RuntimeBeanReference(CHUNK_PROCESSOR_CHUNK_HANDLER_BEAN_NAME_PREFIX + id));
return builder;
}
}
}