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

Sonar: repeated literals

Browse Source 
* Polishing - PR Comments

* GatewayParser: Restore suppress warnings; remove size from `toArray()`.

* Merge conflict resolution
This commit is contained in:
Gary Russell
2019-05-03 12:39:02 -04:00
committed by Artem Bilan
parent 444cd1e8df
commit 6d7bc1fc39
38 changed files with 382 additions and 281 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java
View File

@@ -23,6 +23,7 @@ import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiFunction;
import org.springframework.integration.acks.AcknowledgmentCallback;
import org.springframework.lang.Nullable;
@@ -66,6 +67,9 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
public static final String ACKNOWLEDGMENT_CALLBACK = "acknowledgmentCallback";
private static final BiFunction<String, String, String> TYPE_VERIFY_MESSAGE_FUNCTION =
(name, trailer) -> "The '" + name + trailer;
private Set<String> readOnlyHeaders = new HashSet<>();
public IntegrationMessageHeaderAccessor(@Nullable Message<?> message) {
@@ -168,22 +172,22 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
if (headerName != null && headerValue != null) {
super.verifyType(headerName, headerValue);
if (IntegrationMessageHeaderAccessor.EXPIRATION_DATE.equals(headerName)) {
Assert.isTrue(headerValue instanceof Date || headerValue instanceof Long, "The '" + headerName
+ "' header value must be a Date or Long.");
Assert.isTrue(headerValue instanceof Date || headerValue instanceof Long,
TYPE_VERIFY_MESSAGE_FUNCTION.apply(headerName, "' header value must be a Date or Long."));
}
else if (IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER.equals(headerName)
|| IntegrationMessageHeaderAccessor.SEQUENCE_SIZE.equals(headerName)
|| IntegrationMessageHeaderAccessor.PRIORITY.equals(headerName)) {
Assert.isTrue(Number.class.isAssignableFrom(headerValue.getClass()), "The '" + headerName
+ "' header value must be a Number.");
Assert.isTrue(Number.class.isAssignableFrom(headerValue.getClass()),
TYPE_VERIFY_MESSAGE_FUNCTION.apply(headerName, "' header value must be a Number."));
}
else if (IntegrationMessageHeaderAccessor.ROUTING_SLIP.equals(headerName)) {
Assert.isTrue(Map.class.isAssignableFrom(headerValue.getClass()), "The '" + headerName
+ "' header value must be a Map.");
Assert.isTrue(Map.class.isAssignableFrom(headerValue.getClass()),
TYPE_VERIFY_MESSAGE_FUNCTION.apply(headerName, "' header value must be a Map."));
}
else if (IntegrationMessageHeaderAccessor.DUPLICATE_MESSAGE.equals(headerName)) {
Assert.isTrue(Boolean.class.isAssignableFrom(headerValue.getClass()), "The '" + headerName
+ "' header value must be an Boolean.");
Assert.isTrue(Boolean.class.isAssignableFrom(headerValue.getClass()),
TYPE_VERIFY_MESSAGE_FUNCTION.apply(headerName, "' header value must be an Boolean."));
}
}
}

24
spring-integration-core/src/main/java/org/springframework/integration/config/xml/EnricherParser.java
View File

@@ -47,6 +47,8 @@ import org.springframework.util.xml.DomUtils;
*/
public class EnricherParser extends AbstractConsumerEndpointParser {
private static final String TYPE_ATTRIBUTE = "type";
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
final BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ContentEnricher.class);
@@ -65,8 +67,8 @@ public class EnricherParser extends AbstractConsumerEndpointParser {
String name = subElement.getAttribute("name");
String value = subElement.getAttribute("value");
String type = subElement.getAttribute("type");
String expression = subElement.getAttribute("expression");
String type = subElement.getAttribute(TYPE_ATTRIBUTE);
String expression = subElement.getAttribute(EXPRESSION_ATTRIBUTE);
String nullResultExpression = subElement.getAttribute("null-result-expression");
boolean hasAttributeValue = StringUtils.hasText(value);
boolean hasAttributeExpression = StringUtils.hasText(expression);
@@ -131,13 +133,13 @@ public class EnricherParser extends AbstractConsumerEndpointParser {
String name = subElement.getAttribute("name");
String nullResultHeaderExpression = subElement.getAttribute("null-result-expression");
String valueElementValue = subElement.getAttribute("value");
String expressionElementValue = subElement.getAttribute("expression");
String expressionElementValue = subElement.getAttribute(EXPRESSION_ATTRIBUTE);
boolean hasAttributeValue = StringUtils.hasText(valueElementValue);
boolean hasAttributeExpression = StringUtils.hasText(expressionElementValue);
boolean hasAttributeNullResultExpression = StringUtils.hasText(nullResultHeaderExpression);
if (hasAttributeValue && hasAttributeExpression) {
parserContext.getReaderContext().error("Only one of '" + "value" + "' or '"
+ "expression" + "' is allowed", subElement);
+ EXPRESSION_ATTRIBUTE + "' is allowed", subElement);
}
if (!hasAttributeValue && !hasAttributeExpression && !hasAttributeNullResultExpression) {
@@ -151,11 +153,12 @@ public class EnricherParser extends AbstractConsumerEndpointParser {
}
else if (hasAttributeExpression) {
expressionDef =
IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("expression", subElement);
IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined(EXPRESSION_ATTRIBUTE,
subElement);
}
if (StringUtils.hasText(subElement.getAttribute("expression"))
&& StringUtils.hasText(subElement.getAttribute("type"))) {
if (StringUtils.hasText(subElement.getAttribute(EXPRESSION_ATTRIBUTE))
&& StringUtils.hasText(subElement.getAttribute(TYPE_ATTRIBUTE))) {
parserContext.getReaderContext()
.warning("The use of a 'type' attribute is deprecated since 4.0 "
+ "when using 'expression'", subElement);
@@ -164,8 +167,9 @@ public class EnricherParser extends AbstractConsumerEndpointParser {
BeanDefinitionBuilder valueProcessorBuilder = BeanDefinitionBuilder
.genericBeanDefinition(ExpressionEvaluatingHeaderValueMessageProcessor.class)
.addConstructorArgValue(expressionDef)
.addConstructorArgValue(subElement.getAttribute("type"));
IntegrationNamespaceUtils.setValueIfAttributeDefined(valueProcessorBuilder, subElement, "overwrite");
.addConstructorArgValue(subElement.getAttribute(TYPE_ATTRIBUTE));
IntegrationNamespaceUtils.setValueIfAttributeDefined(valueProcessorBuilder, subElement,
"overwrite");
expressions.put(name, valueProcessorBuilder.getBeanDefinition());
}
if (hasAttributeNullResultExpression) {
@@ -174,7 +178,7 @@ public class EnricherParser extends AbstractConsumerEndpointParser {
BeanDefinitionBuilder nullResultValueProcessorBuilder = BeanDefinitionBuilder
.genericBeanDefinition(ExpressionEvaluatingHeaderValueMessageProcessor.class)
.addConstructorArgValue(nullResultExpressionDefinition)
.addConstructorArgValue(subElement.getAttribute("type"));
.addConstructorArgValue(subElement.getAttribute(TYPE_ATTRIBUTE));
IntegrationNamespaceUtils.setValueIfAttributeDefined(nullResultValueProcessorBuilder, subElement,
"overwrite");
nullResultHeaderExpressions.put(name, nullResultValueProcessorBuilder.getBeanDefinition());

13
spring-integration-core/src/main/java/org/springframework/integration/config/xml/GatewayParser.java
View File

@@ -56,7 +56,8 @@ public class GatewayParser implements BeanDefinitionParser {
boolean isNested = parserContext.isNested();
final Map<String, Object> gatewayAttributes = new HashMap<String, Object>();
gatewayAttributes.put("name", element.getAttribute(AbstractBeanDefinitionParser.ID_ATTRIBUTE));
gatewayAttributes.put(AbstractBeanDefinitionParser.NAME_ATTRIBUTE,
element.getAttribute(AbstractBeanDefinitionParser.ID_ATTRIBUTE));
gatewayAttributes.put("defaultPayloadExpression", element.getAttribute("default-payload-expression"));
gatewayAttributes.put("defaultRequestChannel",
element.getAttribute(isNested ? "request-channel" : "default-request-channel"));
@@ -84,19 +85,20 @@ public class GatewayParser implements BeanDefinitionParser {
List<Map<String, Object>> headers = new ArrayList<Map<String, Object>>(headerElements.size());
for (Element e : headerElements) {
Map<String, Object> header = new HashMap<String, Object>();
header.put("name", e.getAttribute("name"));
header.put(AbstractBeanDefinitionParser.NAME_ATTRIBUTE,
e.getAttribute(AbstractBeanDefinitionParser.NAME_ATTRIBUTE));
header.put("value", e.getAttribute("value"));
header.put("expression", e.getAttribute("expression"));
headers.add(header);
}
gatewayAttributes.put("defaultHeaders", headers.toArray(new Map[headers.size()]));
gatewayAttributes.put("defaultHeaders", headers.toArray(new Map[0]));
}
List<Element> methodElements = DomUtils.getChildElementsByTagName(element, "method");
if (!CollectionUtils.isEmpty(methodElements)) {
Map<String, BeanDefinition> methodMetadataMap = new ManagedMap<String, BeanDefinition>();
for (Element methodElement : methodElements) {
String methodName = methodElement.getAttribute("name");
String methodName = methodElement.getAttribute(AbstractBeanDefinitionParser.NAME_ATTRIBUTE);
BeanDefinitionBuilder methodMetadataBuilder = BeanDefinitionBuilder.genericBeanDefinition(
GatewayMethodMetadata.class);
methodMetadataBuilder.addPropertyValue("requestChannelName",
@@ -122,7 +124,8 @@ public class GatewayParser implements BeanDefinitionParser {
.createExpressionDefinitionFromValueOrExpression("value", "expression", parserContext,
headerElement, true);
headerExpressions.put(headerElement.getAttribute("name"), expressionDef);
headerExpressions.put(headerElement.getAttribute(AbstractBeanDefinitionParser.NAME_ATTRIBUTE),
expressionDef);
}
methodMetadataBuilder.addPropertyValue("headerExpressions", headerExpressions);
}

10
spring-integration-core/src/main/java/org/springframework/integration/config/xml/HeaderEnricherParserSupport.java
View File

@@ -55,6 +55,8 @@ import org.springframework.util.xml.DomUtils;
*/
public abstract class HeaderEnricherParserSupport extends AbstractTransformerParser {
private static final String TYPE_ATTRIBUTE = "type";
private static final Map<String, String[][]> cannedHeaderElementExpressions = new HashMap<>(); // NOSONAR lower case
private final Map<String, String> elementToNameMap = new HashMap<>();
@@ -114,14 +116,14 @@ public abstract class HeaderEnricherParserSupport extends AbstractTransformerPar
else {
headerName = this.elementToNameMap.get(elementName);
headerType = this.elementToTypeMap.get(elementName);
if (headerType != null && StringUtils.hasText(headerElement.getAttribute("type"))) {
if (headerType != null && StringUtils.hasText(headerElement.getAttribute(TYPE_ATTRIBUTE))) {
parserContext.getReaderContext().error("The " + elementName
+ " header does not accept a 'type' attribute. The required type is ["
+ headerType + "]", element);
}
}
if (headerType == null) {
headerType = headerElement.getAttribute("type");
headerType = headerElement.getAttribute(TYPE_ATTRIBUTE);
}
if (headerName == null) {
String ttlExpression = headerElement.getAttribute("time-to-live-expression");
@@ -261,7 +263,7 @@ public abstract class HeaderEnricherParserSupport extends AbstractTransformerPar
valueProcessorBuilder.addConstructorArgValue(headerType);
}
else if (isCustomBean) {
if (StringUtils.hasText(headerElement.getAttribute("type"))) {
if (StringUtils.hasText(headerElement.getAttribute(TYPE_ATTRIBUTE))) {
parserContext.getReaderContext().error(
"The 'type' attribute cannot be used with an inner bean.", element);
}
@@ -280,7 +282,7 @@ public abstract class HeaderEnricherParserSupport extends AbstractTransformerPar
}
}
else {
if (StringUtils.hasText(headerElement.getAttribute("type"))) {
if (StringUtils.hasText(headerElement.getAttribute(TYPE_ATTRIBUTE))) {
parserContext.getReaderContext().error(
"The 'type' attribute cannot be used with the 'ref' attribute.", element);
}

6
spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java
View File

@@ -255,7 +255,7 @@ public abstract class IntegrationNamespaceUtils {
public static void configurePollerMetadata(Element pollerElement, BeanDefinitionBuilder targetBuilder,
ParserContext parserContext) {
if (pollerElement.hasAttribute("ref")) {
if (pollerElement.hasAttribute(REF_ATTRIBUTE)) {
int numberOfAttributes = pollerElement.getAttributes().getLength();
if (numberOfAttributes != 1) {
/*
@@ -273,7 +273,7 @@ public abstract class IntegrationNamespaceUtils {
parserContext.getReaderContext().error(
"A 'poller' element that provides a 'ref' must have no child elements.", pollerElement);
}
targetBuilder.addPropertyReference("pollerMetadata", pollerElement.getAttribute("ref"));
targetBuilder.addPropertyReference("pollerMetadata", pollerElement.getAttribute(REF_ATTRIBUTE));
}
else {
BeanDefinition beanDefinition = parserContext.getDelegate().parseCustomElement(pollerElement,
@@ -519,7 +519,7 @@ public abstract class IntegrationNamespaceUtils {
parserContext.registerBeanComponent(new BeanComponentDefinition(holder)); // NOSONAR never null
adviceChain.add(new RuntimeBeanReference(holder.getBeanName()));
}
else if ("ref".equals(localName)) {
else if (REF_ATTRIBUTE.equals(localName)) {
String ref = childElement.getAttribute("bean");
adviceChain.add(new RuntimeBeanReference(ref));
}

10
spring-integration-core/src/main/java/org/springframework/integration/config/xml/PublishingInterceptorParser.java
View File

@@ -46,6 +46,8 @@ import org.springframework.util.xml.DomUtils;
*/
public class PublishingInterceptorParser extends AbstractBeanDefinitionParser {
private static final String PAYLOAD = "payload";
@Override
protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) {
BeanDefinitionBuilder rootBuilder = BeanDefinitionBuilder.genericBeanDefinition(
@@ -54,7 +56,7 @@ public class PublishingInterceptorParser extends AbstractBeanDefinitionParser {
.genericBeanDefinition(MethodNameMappingPublisherMetadataSource.class);
Map<String, Map<?, ?>> mappings = this
.getMappings(element, element.getAttribute("default-channel"), parserContext);
spelSourceBuilder.addConstructorArgValue(mappings.get("payload"));
spelSourceBuilder.addConstructorArgValue(mappings.get(PAYLOAD));
if (mappings.get("headers") != null) {
spelSourceBuilder.addPropertyValue("headerExpressionMap", mappings.get("headers"));
}
@@ -81,8 +83,8 @@ public class PublishingInterceptorParser extends AbstractBeanDefinitionParser {
// set payloadMap
String methodPattern = StringUtils.hasText(mapping.getAttribute("pattern")) ?
mapping.getAttribute("pattern") : "*";
String payloadExpression = StringUtils.hasText(mapping.getAttribute("payload")) ?
mapping.getAttribute("payload") : "#return";
String payloadExpression = StringUtils.hasText(mapping.getAttribute(PAYLOAD)) ?
mapping.getAttribute(PAYLOAD) : "#return";
payloadExpressionMap.put(methodPattern, payloadExpression);
// set headersMap
@@ -125,7 +127,7 @@ public class PublishingInterceptorParser extends AbstractBeanDefinitionParser {
if (payloadExpressionMap.size() == 0) {
payloadExpressionMap.put("*", "#return");
}
interceptorMappings.put("payload", payloadExpressionMap);
interceptorMappings.put(PAYLOAD, payloadExpressionMap);
if (headersExpressionMap.size() > 0) {
interceptorMappings.put("headers", headersExpressionMap);
}

10
spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResourceInboundChannelAdapterParser.java
View File

@@ -29,26 +29,28 @@ import org.springframework.util.StringUtils;
* Parser for 'resource-inbound-channel-adapter'
*
* @author Oleg Zhurakousky
* @author Gary Russell
* @since 2.1
*/
public class ResourceInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
private static final String FILTER = "filter";
@Override
protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
BeanDefinitionBuilder sourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(ResourceRetrievingMessageSource.class);
sourceBuilder.addConstructorArgValue(element.getAttribute("pattern"));
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(sourceBuilder, element, "pattern-resolver");
boolean hasFilter = element.hasAttribute("filter");
boolean hasFilter = element.hasAttribute(FILTER);
if (hasFilter) {
String filterValue = element.getAttribute("filter");
String filterValue = element.getAttribute(FILTER);
if (StringUtils.hasText(filterValue)) {
sourceBuilder.addPropertyReference("filter", filterValue);
sourceBuilder.addPropertyReference(FILTER, filterValue);
}
}
else {
BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition(AcceptOnceCollectionFilter.class);
sourceBuilder.addPropertyValue("filter", filterBuilder.getBeanDefinition());
sourceBuilder.addPropertyValue(FILTER, filterBuilder.getBeanDefinition());
}
return sourceBuilder.getBeanDefinition();
}

20
spring-integration-core/src/main/java/org/springframework/integration/core/AsyncMessagingTemplate.java
View File

@@ -35,6 +35,8 @@ import org.springframework.util.Assert;
*/
public class AsyncMessagingTemplate extends MessagingTemplate implements AsyncMessagingOperations {
private static final String UNCHECKED = "unchecked";
private volatile AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor();
@@ -90,19 +92,19 @@ public class AsyncMessagingTemplate extends MessagingTemplate implements AsyncMe
}
@Override
@SuppressWarnings("unchecked")
@SuppressWarnings(UNCHECKED)
public <R> Future<R> asyncReceiveAndConvert() {
return this.executor.submit(() -> (R) receiveAndConvert(Object.class));
}
@Override
@SuppressWarnings("unchecked")
@SuppressWarnings(UNCHECKED)
public <R> Future<R> asyncReceiveAndConvert(final PollableChannel channel) {
return this.executor.submit(() -> (R) receiveAndConvert(channel, Object.class));
}
@Override
@SuppressWarnings("unchecked")
@SuppressWarnings(UNCHECKED)
public <R> Future<R> asyncReceiveAndConvert(final String channelName) {
return this.executor.submit(() -> (R) receiveAndConvert(channelName, Object.class));
}
@@ -123,32 +125,32 @@ public class AsyncMessagingTemplate extends MessagingTemplate implements AsyncMe
}
@Override
@SuppressWarnings("unchecked")
@SuppressWarnings(UNCHECKED)
public <R> Future<R> asyncConvertSendAndReceive(final Object request) {
return this.executor.submit(() -> (R) convertSendAndReceive(request, Object.class));
}
@Override
@SuppressWarnings("unchecked")
@SuppressWarnings(UNCHECKED)
public <R> Future<R> asyncConvertSendAndReceive(final MessageChannel channel, final Object request) {
return this.executor.submit(() -> (R) convertSendAndReceive(channel, request, Object.class));
}
@Override
@SuppressWarnings("unchecked")
@SuppressWarnings(UNCHECKED)
public <R> Future<R> asyncConvertSendAndReceive(final String channelName, final Object request) {
return this.executor.submit(() -> (R) convertSendAndReceive(channelName, request, Object.class));
}
@Override
@SuppressWarnings("unchecked")
@SuppressWarnings(UNCHECKED)
public <R> Future<R> asyncConvertSendAndReceive(final Object request,
final MessagePostProcessor requestPostProcessor) {
return this.executor.submit(() -> (R) convertSendAndReceive(request, Object.class, requestPostProcessor));
}
@Override
@SuppressWarnings("unchecked")
@SuppressWarnings(UNCHECKED)
public <R> Future<R> asyncConvertSendAndReceive(final MessageChannel channel, final Object request,
final MessagePostProcessor requestPostProcessor) {
return this.executor
@@ -156,7 +158,7 @@ public class AsyncMessagingTemplate extends MessagingTemplate implements AsyncMe
}
@Override
@SuppressWarnings("unchecked")
@SuppressWarnings(UNCHECKED)
public <R> Future<R> asyncConvertSendAndReceive(final String channelName, final Object request,
final MessagePostProcessor requestPostProcessor) {
return this.executor

10
spring-integration-core/src/main/java/org/springframework/integration/dsl/HeaderEnricherSpec.java
View File

@@ -56,6 +56,8 @@ import reactor.util.function.Tuple2;
*/
public class HeaderEnricherSpec extends ConsumerEndpointSpec<HeaderEnricherSpec, MessageTransformingHandler> {
private static final String HEADERS_MUST_NOT_BE_NULL = "'headers' must not be null";
private final Map<String, HeaderValueMessageProcessor<?>> headerToAdd = new HashMap<>();
private final HeaderEnricher headerEnricher = new HeaderEnricher(this.headerToAdd);
@@ -150,7 +152,7 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec<HeaderEnricherSpec,
* @return the header enricher spec.
*/
public HeaderEnricherSpec headers(MapBuilder<?, String, Object> headers, Boolean overwrite) {
Assert.notNull(headers, "'headers' must not be null");
Assert.notNull(headers, HEADERS_MUST_NOT_BE_NULL);
return headers(headers.get(), overwrite);
}
@@ -176,7 +178,7 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec<HeaderEnricherSpec,
* @return the header enricher spec.
*/
public HeaderEnricherSpec headers(Map<String, Object> headers, Boolean overwrite) {
Assert.notNull(headers, "'headers' must not be null");
Assert.notNull(headers, HEADERS_MUST_NOT_BE_NULL);
for (Entry<String, Object> entry : headers.entrySet()) {
String name = entry.getKey();
Object value = entry.getValue();
@@ -214,7 +216,7 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec<HeaderEnricherSpec,
* @return the header enricher spec.
*/
public HeaderEnricherSpec headerExpressions(MapBuilder<?, String, String> headers, Boolean overwrite) {
Assert.notNull(headers, "'headers' must not be null");
Assert.notNull(headers, HEADERS_MUST_NOT_BE_NULL);
return headerExpressions(headers.get(), overwrite);
}
@@ -283,7 +285,7 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec<HeaderEnricherSpec,
* @return the header enricher spec.
*/
public HeaderEnricherSpec headerExpressions(Map<String, String> headers, Boolean overwrite) {
Assert.notNull(headers, "'headers' must not be null");
Assert.notNull(headers, HEADERS_MUST_NOT_BE_NULL);
for (Entry<String, String> entry : headers.entrySet()) {
AbstractHeaderValueMessageProcessor<Object> processor =
new ExpressionEvaluatingHeaderValueMessageProcessor<>(entry.getValue(), null);

32
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
View File

@@ -115,6 +115,12 @@ import reactor.util.function.Tuple2;
*/
public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinition<B>> {
private static final String UNCHECKED = "unchecked";
private static final String FUNCTION_MUST_NOT_BE_NULL = "'function' must not be null";
private static final String MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL = "'messageProcessorSpec' must not be null";
private static final SpelExpressionParser PARSER = new SpelExpressionParser();
private static final Set<MessageProducer> REFERENCED_REPLY_PRODUCERS = new HashSet<>();
@@ -582,7 +588,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
*/
public B transform(MessageProcessorSpec<?> messageProcessorSpec,
Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
Assert.notNull(messageProcessorSpec, "'messageProcessorSpec' must not be null");
Assert.notNull(messageProcessorSpec, MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL);
MessageProcessor<?> processor = messageProcessorSpec.get();
return addComponent(processor)
.transform(new MethodInvokingTransformer(processor), endpointConfigurer);
@@ -814,7 +820,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @return the current {@link IntegrationFlowDefinition}.
*/
public B filter(MessageProcessorSpec<?> messageProcessorSpec, Consumer<FilterEndpointSpec> endpointConfigurer) {
Assert.notNull(messageProcessorSpec, "'messageProcessorSpec' must not be null");
Assert.notNull(messageProcessorSpec, MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL);
MessageProcessor<?> processor = messageProcessorSpec.get();
return addComponent(processor)
.filter(new MethodInvokingSelector(processor), endpointConfigurer);
@@ -1133,7 +1139,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
*/
public B handle(MessageProcessorSpec<?> messageProcessorSpec,
Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
Assert.notNull(messageProcessorSpec, "'messageProcessorSpec' must not be null");
Assert.notNull(messageProcessorSpec, MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL);
MessageProcessor<?> processor = messageProcessorSpec.get();
return addComponent(processor)
.handle(new ServiceActivatingHandler(processor), endpointConfigurer);
@@ -1518,7 +1524,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
*/
public B split(MessageProcessorSpec<?> messageProcessorSpec,
Consumer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
Assert.notNull(messageProcessorSpec, "'messageProcessorSpec' must not be null");
Assert.notNull(messageProcessorSpec, MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL);
MessageProcessor<?> processor = messageProcessorSpec.get();
return addComponent(processor)
.split(new MethodInvokingSplitter(processor), endpointConfigurer);
@@ -2056,7 +2062,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
*/
public B route(MessageProcessorSpec<?> messageProcessorSpec,
Consumer<RouterSpec<Object, MethodInvokingRouter>> routerConfigurer) {
Assert.notNull(messageProcessorSpec, "'messageProcessorSpec' must not be null");
Assert.notNull(messageProcessorSpec, MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL);
MessageProcessor<?> processor = messageProcessorSpec.get();
addComponent(processor);
@@ -2371,7 +2377,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @see #wireTap(WireTapSpec)
*/
public <P> B log(Function<Message<P>, Object> function) {
Assert.notNull(function, "'function' must not be null");
Assert.notNull(function, FUNCTION_MUST_NOT_BE_NULL);
return log(new FunctionExpression<>(function));
}
@@ -2488,7 +2494,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @see #wireTap(WireTapSpec)
*/
public <P> B log(LoggingHandler.Level level, String category, Function<Message<P>, Object> function) {
Assert.notNull(function, "'function' must not be null");
Assert.notNull(function, FUNCTION_MUST_NOT_BE_NULL);
return log(level, category, new FunctionExpression<>(function));
}
@@ -2632,7 +2638,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @see #bridge()
*/
public <P> IntegrationFlow logAndReply(Function<Message<P>, Object> function) {
Assert.notNull(function, "'function' must not be null");
Assert.notNull(function, FUNCTION_MUST_NOT_BE_NULL);
return logAndReply(new FunctionExpression<>(function));
}
@@ -2757,7 +2763,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
public <P> IntegrationFlow logAndReply(LoggingHandler.Level level, String category,
Function<Message<P>, Object> function) {
Assert.notNull(function, "'function' must not be null");
Assert.notNull(function, FUNCTION_MUST_NOT_BE_NULL);
return logAndReply(level, category, new FunctionExpression<>(function));
}
@@ -2961,7 +2967,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @param <O> the output type.
* @return the current {@link IntegrationFlowDefinition}.
*/
@SuppressWarnings("unchecked")
@SuppressWarnings(UNCHECKED)
public <I, O> B fluxTransform(Function<? super Flux<Message<I>>, ? extends Publisher<O>> fluxFunction) {
if (!(this.currentMessageChannel instanceof FluxMessageChannel)) {
channel(new FluxMessageChannel());
@@ -2984,7 +2990,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @param <T> the expected {@code payload} type
* @return the Reactive Streams {@link Publisher}
*/
@SuppressWarnings("unchecked")
@SuppressWarnings(UNCHECKED)
public <T> Publisher<Message<T>> toReactivePublisher() {
MessageChannel channelForPublisher = this.currentMessageChannel;
Publisher<Message<T>> publisher;
@@ -3022,7 +3028,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
.get();
}
@SuppressWarnings("unchecked")
@SuppressWarnings(UNCHECKED)
private <S extends ConsumerEndpointSpec<S, ? extends MessageHandler>> B register(S endpointSpec,
Consumer<S> endpointConfigurer) {
@@ -3113,7 +3119,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
return false;
}
@SuppressWarnings("unchecked")
@SuppressWarnings(UNCHECKED)
protected final B _this() { // NOSONAR name
return (B) this;
}

23
spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistory.java
View File

@@ -52,6 +52,9 @@ public final class MessageHistory implements List<Properties>, Serializable {
private static final Log logger = LogFactory.getLog(MessageHistory.class);
private static final UnsupportedOperationException UNSUPPORTED_OPERATION_EXCEPTION_IMMUTABLE =
new UnsupportedOperationException("MessageHistory is immutable.");
public static final String HEADER_NAME = "history";
public static final String NAME_PROPERTY = "name";
@@ -215,52 +218,52 @@ public final class MessageHistory implements List<Properties>, Serializable {
@Override
public boolean add(Properties e) {
throw new UnsupportedOperationException("MessageHistory is immutable.");
throw UNSUPPORTED_OPERATION_EXCEPTION_IMMUTABLE;
}
@Override
public void add(int index, Properties element) {
throw new UnsupportedOperationException("MessageHistory is immutable.");
throw UNSUPPORTED_OPERATION_EXCEPTION_IMMUTABLE;
}
@Override
public boolean addAll(Collection<? extends Properties> c) {
throw new UnsupportedOperationException("MessageHistory is immutable.");
throw UNSUPPORTED_OPERATION_EXCEPTION_IMMUTABLE;
}
@Override
public boolean addAll(int index, Collection<? extends Properties> c) {
throw new UnsupportedOperationException("MessageHistory is immutable.");
throw UNSUPPORTED_OPERATION_EXCEPTION_IMMUTABLE;
}
@Override
public Properties set(int index, Properties element) {
throw new UnsupportedOperationException("MessageHistory is immutable.");
throw UNSUPPORTED_OPERATION_EXCEPTION_IMMUTABLE;
}
@Override
public Properties remove(int index) {
throw new UnsupportedOperationException("MessageHistory is immutable.");
throw UNSUPPORTED_OPERATION_EXCEPTION_IMMUTABLE;
}
@Override
public boolean remove(Object o) {
throw new UnsupportedOperationException("MessageHistory is immutable.");
throw UNSUPPORTED_OPERATION_EXCEPTION_IMMUTABLE;
}
@Override
public boolean removeAll(Collection<?> c) {
throw new UnsupportedOperationException("MessageHistory is immutable.");
throw UNSUPPORTED_OPERATION_EXCEPTION_IMMUTABLE;
}
@Override
public boolean retainAll(Collection<?> c) {
throw new UnsupportedOperationException("MessageHistory is immutable.");
throw UNSUPPORTED_OPERATION_EXCEPTION_IMMUTABLE;
}
@Override
public void clear() {
throw new UnsupportedOperationException("MessageHistory is immutable.");
throw UNSUPPORTED_OPERATION_EXCEPTION_IMMUTABLE;
}

12
spring-integration-core/src/main/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStore.java
View File

@@ -55,6 +55,8 @@ import org.springframework.util.DefaultPropertiesPersister;
public class PropertiesPersistingMetadataStore implements ConcurrentMetadataStore, InitializingBean, DisposableBean,
Closeable, Flushable {
private static final String KEY_CANNOT_BE_NULL = "'key' cannot be null";
private final Log logger = LogFactory.getLog(getClass());
private final Properties metadata = new Properties();
@@ -111,7 +113,7 @@ public class PropertiesPersistingMetadataStore implements ConcurrentMetadataStor
@Override
public void put(String key, String value) {
Assert.notNull(key, "'key' cannot be null");
Assert.notNull(key, KEY_CANNOT_BE_NULL);
Assert.notNull(value, "'value' cannot be null");
Lock lock = this.lockRegistry.obtain(key);
lock.lock();
@@ -126,7 +128,7 @@ public class PropertiesPersistingMetadataStore implements ConcurrentMetadataStor
@Override
public String get(String key) {
Assert.notNull(key, "'key' cannot be null");
Assert.notNull(key, KEY_CANNOT_BE_NULL);
Lock lock = this.lockRegistry.obtain(key);
lock.lock();
try {
@@ -139,7 +141,7 @@ public class PropertiesPersistingMetadataStore implements ConcurrentMetadataStor
@Override
public String remove(String key) {
Assert.notNull(key, "'key' cannot be null");
Assert.notNull(key, KEY_CANNOT_BE_NULL);
Lock lock = this.lockRegistry.obtain(key);
lock.lock();
try {
@@ -153,7 +155,7 @@ public class PropertiesPersistingMetadataStore implements ConcurrentMetadataStor
@Override
public String putIfAbsent(String key, String value) {
Assert.notNull(key, "'key' cannot be null");
Assert.notNull(key, KEY_CANNOT_BE_NULL);
Assert.notNull(value, "'value' cannot be null");
Lock lock = this.lockRegistry.obtain(key);
lock.lock();
@@ -175,7 +177,7 @@ public class PropertiesPersistingMetadataStore implements ConcurrentMetadataStor
@Override
public boolean replace(String key, String oldValue, String newValue) {
Assert.notNull(key, "'key' cannot be null");
Assert.notNull(key, KEY_CANNOT_BE_NULL);
Assert.notNull(oldValue, "'oldValue' cannot be null");
Assert.notNull(newValue, "'newValue' cannot be null");
Lock lock = this.lockRegistry.obtain(key);

14
spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java
View File

@@ -40,6 +40,8 @@ import org.springframework.util.Assert;
*/
public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupStore implements MessageStore {
private static final String GROUP_ID_MUST_NOT_BE_NULL = "'groupId' must not be null";
protected static final String MESSAGE_KEY_PREFIX = "MESSAGE_";
protected static final String MESSAGE_GROUP_KEY_PREFIX = "MESSAGE_GROUP_";
@@ -193,7 +195,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
@Override
public MessageGroupMetadata getGroupMetadata(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL);
Object mgm = this.doRetrieve(this.groupPrefix + groupId);
if (mgm != null) {
Assert.isInstanceOf(MessageGroupMetadata.class, mgm);
@@ -204,7 +206,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
@Override
public void addMessagesToGroup(Object groupId, Message<?>... messages) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL);
Assert.notNull(messages, "'messages' must not be null");
MessageGroupMetadata metadata = getGroupMetadata(groupId);
@@ -239,7 +241,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
@Override
public void removeMessagesFromGroup(Object groupId, Collection<Message<?>> messages) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL);
Assert.notNull(messages, "'messages' must not be null");
Object mgm = doRetrieve(this.groupPrefix + groupId);
@@ -268,7 +270,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
@Override
public void completeGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL);
MessageGroupMetadata metadata = getGroupMetadata(groupId);
if (metadata != null) {
metadata.complete();
@@ -282,7 +284,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
*/
@Override
public void removeMessageGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL);
Object mgm = doRemove(this.groupPrefix + groupId);
if (mgm != null) {
Assert.isInstanceOf(MessageGroupMetadata.class, mgm);
@@ -300,7 +302,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
@Override
public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL);
MessageGroupMetadata metadata = getGroupMetadata(groupId);
if (metadata == null) {
SimpleMessageGroup messageGroup = new SimpleMessageGroup(groupId);

36
spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java
View File

@@ -50,6 +50,12 @@ import org.springframework.util.CollectionUtils;
public class SimpleMessageStore extends AbstractMessageGroupStore
implements MessageStore, ChannelMessageStore {
private static final String MESSAGE_GROUP_FOR_GROUP_ID = "MessageGroup for groupId '";
private static final String UPPER_BOUND_MUST_NOT_BE_NULL = "'upperBound' must not be null.";
private static final String INTERRUPTED_WHILE_OBTAINING_LOCK = "Interrupted while obtaining lock";
private final ConcurrentMap<UUID, Message<?>> idToMessage = new ConcurrentHashMap<UUID, Message<?>>();
private final ConcurrentMap<Object, MessageGroup> groupIdToMessageGroup =
@@ -256,7 +262,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new MessagingException("Interrupted while obtaining lock", e);
throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e);
}
}
@@ -291,7 +297,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore
}
else {
upperBound = this.groupToUpperBound.get(groupId);
Assert.state(upperBound != null, "'upperBound' must not be null.");
Assert.state(upperBound != null, UPPER_BOUND_MUST_NOT_BE_NULL);
for (Message<?> message : messages) {
lock.unlock();
if (!upperBound.tryAcquire(this.upperBoundTimeout)) {
@@ -313,7 +319,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new MessagingException("Interrupted while obtaining lock", e);
throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e);
}
}
@@ -326,7 +332,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore
MessageGroup messageGroup = this.groupIdToMessageGroup.remove(groupId);
if (messageGroup != null) {
UpperBound upperBound = this.groupToUpperBound.remove(groupId);
Assert.state(upperBound != null, "'upperBound' must not be null.");
Assert.state(upperBound != null, UPPER_BOUND_MUST_NOT_BE_NULL);
upperBound.release(this.groupCapacity);
}
}
@@ -336,7 +342,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new MessagingException("Interrupted while obtaining lock", e);
throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e);
}
}
@@ -347,10 +353,10 @@ public class SimpleMessageStore extends AbstractMessageGroupStore
lock.lockInterruptibly();
try {
MessageGroup group = this.groupIdToMessageGroup.get(groupId);
Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " +
Assert.notNull(group, MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " +
"can not be located while attempting to remove Message(s) from the MessageGroup");
UpperBound upperBound = this.groupToUpperBound.get(groupId);
Assert.state(upperBound != null, "'upperBound' must not be null.");
Assert.state(upperBound != null, UPPER_BOUND_MUST_NOT_BE_NULL);
boolean modified = false;
for (Message<?> messageToRemove : messages) {
if (group.remove(messageToRemove)) {
@@ -368,7 +374,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new MessagingException("Interrupted while obtaining lock", e);
throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e);
}
}
@@ -384,7 +390,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore
lock.lockInterruptibly();
try {
MessageGroup group = this.groupIdToMessageGroup.get(groupId);
Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " +
Assert.notNull(group, MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " +
"can not be located while attempting to set 'lastReleasedSequenceNumber'");
group.setLastReleasedMessageSequenceNumber(sequenceNumber);
group.setLastModified(System.currentTimeMillis());
@@ -395,7 +401,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new MessagingException("Interrupted while obtaining lock", e);
throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e);
}
}
@@ -406,7 +412,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore
lock.lockInterruptibly();
try {
MessageGroup group = this.groupIdToMessageGroup.get(groupId);
Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " +
Assert.notNull(group, MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " +
"can not be located while attempting to complete the MessageGroup");
group.complete();
group.setLastModified(System.currentTimeMillis());
@@ -417,7 +423,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new MessagingException("Interrupted while obtaining lock", e);
throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e);
}
}
@@ -460,12 +466,12 @@ public class SimpleMessageStore extends AbstractMessageGroupStore
lock.lockInterruptibly();
try {
MessageGroup group = this.groupIdToMessageGroup.get(groupId);
Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " +
Assert.notNull(group, MESSAGE_GROUP_FOR_GROUP_ID + groupId + "' " +
"can not be located while attempting to complete the MessageGroup");
group.clear();
group.setLastModified(System.currentTimeMillis());
UpperBound upperBound = this.groupToUpperBound.get(groupId);
Assert.state(upperBound != null, "'upperBound' must not be null.");
Assert.state(upperBound != null, UPPER_BOUND_MUST_NOT_BE_NULL);
upperBound.release(this.groupCapacity);
}
finally {
@@ -474,7 +480,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new MessagingException("Interrupted while obtaining lock", e);
throw new MessagingException(INTERRUPTED_WHILE_OBTAINING_LOCK, e);
}
}

10
spring-integration-core/src/main/java/org/springframework/integration/support/SmartLifecycleRoleController.java
View File

@@ -59,6 +59,8 @@ import org.springframework.util.MultiValueMap;
public class SmartLifecycleRoleController implements ApplicationListener<AbstractLeaderEvent>,
ApplicationContextAware {
private static final String IN_ROLE = " in role ";
private static final Log logger = LogFactory.getLog(SmartLifecycleRoleController.class);
private final MultiValueMap<String, SmartLifecycle> lifecycles = new LinkedMultiValueMap<String, SmartLifecycle>();
@@ -162,7 +164,7 @@ public class SmartLifecycleRoleController implements ApplicationListener<Abstrac
componentsInRole = new ArrayList<>(componentsInRole);
componentsInRole.sort(Comparator.comparingInt(Phased::getPhase));
if (logger.isDebugEnabled()) {
logger.debug("Starting " + componentsInRole + " in role " + role);
logger.debug("Starting " + componentsInRole + IN_ROLE + role);
}
componentsInRole.forEach(lifecycle -> {
@@ -170,7 +172,7 @@ public class SmartLifecycleRoleController implements ApplicationListener<Abstrac
lifecycle.start();
}
catch (Exception e) {
logger.error("Failed to start " + lifecycle + " in role " + role, e);
logger.error("Failed to start " + lifecycle + IN_ROLE + role, e);
}
});
}
@@ -194,7 +196,7 @@ public class SmartLifecycleRoleController implements ApplicationListener<Abstrac
componentsInRole = new ArrayList<>(componentsInRole);
componentsInRole.sort((o1, o2) -> Integer.compare(o2.getPhase(), o1.getPhase()));
if (logger.isDebugEnabled()) {
logger.debug("Stopping " + componentsInRole + " in role " + role);
logger.debug("Stopping " + componentsInRole + IN_ROLE + role);
}
componentsInRole.forEach(lifecycle -> {
@@ -202,7 +204,7 @@ public class SmartLifecycleRoleController implements ApplicationListener<Abstrac
lifecycle.stop();
}
catch (Exception e) {
logger.error("Failed to stop " + lifecycle + " in role " + role, e);
logger.error("Failed to stop " + lifecycle + IN_ROLE + role, e);
}
});
}

23
spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionSynchronizationFactoryBean.java
View File

@@ -43,6 +43,9 @@ import org.springframework.util.StringUtils;
public class TransactionSynchronizationFactoryBean implements FactoryBean<DefaultTransactionSynchronizationFactory>,
BeanFactoryAware {
private static final String EXPRESSION_OR_CHANNEL_NEEDED =
"At least one attribute ('expression' and/or 'messageChannel') must be defined";
private static final SpelExpressionParser PARSER = new SpelExpressionParser();
private final AtomicInteger counter = new AtomicInteger();
@@ -77,13 +80,13 @@ public class TransactionSynchronizationFactoryBean implements FactoryBean<Defaul
/**
* Specify the {@link DestinationResolver} strategy to use.
* The default is a BeanFactoryChannelResolver.
* @param channelResolver The channel resolver.
* @param resolver The channel resolver.
* @return current TransactionSynchronizationFactoryBean
* @since 4.1.3
*/
public TransactionSynchronizationFactoryBean channelResolver(DestinationResolver<MessageChannel> channelResolver) {
Assert.notNull(channelResolver, "'channelResolver' must not be null");
this.channelResolver = channelResolver;
public TransactionSynchronizationFactoryBean channelResolver(DestinationResolver<MessageChannel> resolver) {
Assert.notNull(resolver, "'channelResolver' must not be null");
this.channelResolver = resolver;
return this;
}
@@ -94,7 +97,7 @@ public class TransactionSynchronizationFactoryBean implements FactoryBean<Defaul
public TransactionSynchronizationFactoryBean beforeCommit(String expression, String messageChannel) {
Assert.state(StringUtils.hasText(expression) || StringUtils.hasText(messageChannel),
"At least one attribute ('expression' and/or 'messageChannel') must be defined");
EXPRESSION_OR_CHANNEL_NEEDED);
this.beforeCommitExpression = expression;
this.beforeCommitChannelName = messageChannel;
this.beforeCommitChannel = null;
@@ -107,7 +110,7 @@ public class TransactionSynchronizationFactoryBean implements FactoryBean<Defaul
public TransactionSynchronizationFactoryBean beforeCommit(String expression, MessageChannel messageChannel) {
Assert.state(StringUtils.hasText(expression) || messageChannel != null,
"At least one attribute ('expression' and/or 'messageChannel') must be defined");
EXPRESSION_OR_CHANNEL_NEEDED);
this.beforeCommitExpression = expression;
this.beforeCommitChannel = messageChannel;
this.beforeCommitChannelName = null;
@@ -120,7 +123,7 @@ public class TransactionSynchronizationFactoryBean implements FactoryBean<Defaul
public TransactionSynchronizationFactoryBean afterCommit(String expression, String messageChannel) {
Assert.state(StringUtils.hasText(expression) || StringUtils.hasText(messageChannel),
"At least one attribute ('expression' and/or 'messageChannel') must be defined");
EXPRESSION_OR_CHANNEL_NEEDED);
this.afterCommitExpression = expression;
this.afterCommitChannelName = messageChannel;
this.afterCommitChannel = null;
@@ -133,7 +136,7 @@ public class TransactionSynchronizationFactoryBean implements FactoryBean<Defaul
public TransactionSynchronizationFactoryBean afterCommit(String expression, MessageChannel messageChannel) {
Assert.state(StringUtils.hasText(expression) || messageChannel != null,
"At least one attribute ('expression' and/or 'messageChannel') must be defined");
EXPRESSION_OR_CHANNEL_NEEDED);
this.afterCommitExpression = expression;
this.afterCommitChannel = messageChannel;
this.afterCommitChannelName = null;
@@ -146,7 +149,7 @@ public class TransactionSynchronizationFactoryBean implements FactoryBean<Defaul
public TransactionSynchronizationFactoryBean afterRollback(String expression, String messageChannel) {
Assert.state(StringUtils.hasText(expression) || StringUtils.hasText(messageChannel),
"At least one attribute ('expression' and/or 'messageChannel') must be defined");
EXPRESSION_OR_CHANNEL_NEEDED);
this.afterRollbackExpression = expression;
this.afterRollbackChannelName = messageChannel;
this.afterRollbackChannel = null;
@@ -159,7 +162,7 @@ public class TransactionSynchronizationFactoryBean implements FactoryBean<Defaul
public TransactionSynchronizationFactoryBean afterRollback(String expression, MessageChannel messageChannel) {
Assert.state(StringUtils.hasText(expression) || messageChannel != null,
"At least one attribute ('expression' and/or 'messageChannel') must be defined");
EXPRESSION_OR_CHANNEL_NEEDED);
this.afterRollbackExpression = expression;
this.afterRollbackChannel = messageChannel;
this.afterRollbackChannelName = null;

12
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java
View File

@@ -42,6 +42,8 @@ import org.springframework.util.ObjectUtils;
*/
public class FtpSession implements Session<FTPFile> {
private static final String SERVER_REPLIED_WITH = "'. Server replied with: ";
private final Log logger = LogFactory.getLog(this.getClass());
private final FTPClient client;
@@ -58,7 +60,7 @@ public class FtpSession implements Session<FTPFile> {
public boolean remove(String path) throws IOException {
Assert.hasText(path, "path must not be null");
if (!this.client.deleteFile(path)) {
throw new IOException("Failed to delete '" + path + "'. Server replied with: " + this.client.getReplyString());
throw new IOException("Failed to delete '" + path + SERVER_REPLIED_WITH + this.client.getReplyString());
}
else {
return true;
@@ -82,7 +84,7 @@ public class FtpSession implements Session<FTPFile> {
boolean completed = this.client.retrieveFile(path, fos);
if (!completed) {
throw new IOException("Failed to copy '" + path +
"'. Server replied with: " + this.client.getReplyString());
SERVER_REPLIED_WITH + this.client.getReplyString());
}
this.logger.info("File has been successfully transferred from: " + path);
}
@@ -122,7 +124,7 @@ public class FtpSession implements Session<FTPFile> {
boolean completed = this.client.storeFile(path, inputStream);
if (!completed) {
throw new IOException("Failed to write to '" + path
+ "'. Server replied with: " + this.client.getReplyString());
+ SERVER_REPLIED_WITH + this.client.getReplyString());
}
if (this.logger.isInfoEnabled()) {
this.logger.info("File has been successfully transferred to: " + path);
@@ -136,7 +138,7 @@ public class FtpSession implements Session<FTPFile> {
boolean completed = this.client.appendFile(path, inputStream);
if (!completed) {
throw new IOException("Failed to append to '" + path
+ "'. Server replied with: " + this.client.getReplyString());
+ SERVER_REPLIED_WITH + this.client.getReplyString());
}
if (this.logger.isInfoEnabled()) {
this.logger.info("File has been successfully appended to: " + path);
@@ -179,7 +181,7 @@ public class FtpSession implements Session<FTPFile> {
boolean completed = this.client.rename(pathFrom, pathTo);
if (!completed) {
throw new IOException("Failed to rename '" + pathFrom +
"' to " + pathTo + "'. Server replied with: " + this.client.getReplyString());
"' to " + pathTo + SERVER_REPLIED_WITH + this.client.getReplyString());
}
if (this.logger.isInfoEnabled()) {
this.logger.info("File has been successfully renamed from: " + pathFrom + " to " + pathTo);

12
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/metadata/GemfireMetadataStore.java
View File

@@ -44,6 +44,8 @@ import org.springframework.util.Assert;
*/
public class GemfireMetadataStore implements ListenableMetadataStore {
private static final String KEY_MUST_NOT_BE_NULL = "'key' must not be null.";
public static final String KEY = "MetaData";
private final GemfireCacheListener cacheListener = new GemfireCacheListener();
@@ -66,21 +68,21 @@ public class GemfireMetadataStore implements ListenableMetadataStore {
@Override
public void put(String key, String value) {
Assert.notNull(key, "'key' must not be null.");
Assert.notNull(key, KEY_MUST_NOT_BE_NULL);
Assert.notNull(value, "'value' must not be null.");
this.region.put(key, value);
}
@Override
public String putIfAbsent(String key, String value) {
Assert.notNull(key, "'key' must not be null.");
Assert.notNull(key, KEY_MUST_NOT_BE_NULL);
Assert.notNull(value, "'value' must not be null.");
return this.region.putIfAbsent(key, value);
}
@Override
public boolean replace(String key, String oldValue, String newValue) {
Assert.notNull(key, "'key' must not be null.");
Assert.notNull(key, KEY_MUST_NOT_BE_NULL);
Assert.notNull(oldValue, "'oldValue' must not be null.");
Assert.notNull(newValue, "'newValue' must not be null.");
return this.region.replace(key, oldValue, newValue);
@@ -88,13 +90,13 @@ public class GemfireMetadataStore implements ListenableMetadataStore {
@Override
public String get(String key) {
Assert.notNull(key, "'key' must not be null.");
Assert.notNull(key, KEY_MUST_NOT_BE_NULL);
return this.region.get(key);
}
@Override
public String remove(String key) {
Assert.notNull(key, "'key' must not be null.");
Assert.notNull(key, KEY_MUST_NOT_BE_NULL);
return this.region.remove(key);
}

10
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java
View File

@@ -41,6 +41,8 @@ import org.springframework.util.PatternMatchUtils;
*/
public class GemfireMessageStore extends AbstractKeyValueMessageStore {
private static final String ID_MUST_NOT_BE_NULL = "'id' must not be null";
private final Region<Object, Object> messageStoreRegion;
/**
@@ -68,20 +70,20 @@ public class GemfireMessageStore extends AbstractKeyValueMessageStore {
@Override
protected Object doRetrieve(Object id) {
Assert.notNull(id, "'id' must not be null");
Assert.notNull(id, ID_MUST_NOT_BE_NULL);
return this.messageStoreRegion.get(id);
}
@Override
protected void doStore(Object id, Object objectToStore) {
Assert.notNull(id, "'id' must not be null");
Assert.notNull(id, ID_MUST_NOT_BE_NULL);
Assert.notNull(objectToStore, "'objectToStore' must not be null");
this.messageStoreRegion.put(id, objectToStore);
}
@Override
protected void doStoreIfAbsent(Object id, Object objectToStore) {
Assert.notNull(id, "'id' must not be null");
Assert.notNull(id, ID_MUST_NOT_BE_NULL);
Assert.notNull(objectToStore, "'objectToStore' must not be null");
Object present = this.messageStoreRegion.putIfAbsent(id, objectToStore);
if (present != null && logger.isDebugEnabled()) {
@@ -92,7 +94,7 @@ public class GemfireMessageStore extends AbstractKeyValueMessageStore {
@Override
protected Object doRemove(Object id) {
Assert.notNull(id, "'id' must not be null");
Assert.notNull(id, ID_MUST_NOT_BE_NULL);
return this.messageStoreRegion.remove(id);
}

14
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/StoredProcParserUtils.java
View File

@@ -30,6 +30,7 @@ import org.springframework.beans.factory.config.TypedStringValue;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.beans.factory.support.ManagedMap;
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.config.ExpressionFactoryBean;
@@ -65,12 +66,13 @@ public final class StoredProcParserUtils {
*/
public static ManagedList<BeanDefinition> getSqlParameterDefinitionBeanDefinitions(
Element storedProcComponent, ParserContext parserContext) {
List<Element> sqlParameterDefinitionChildElements = DomUtils.getChildElementsByTagName(storedProcComponent, "sql-parameter-definition");
List<Element> sqlParameterDefinitionChildElements =
DomUtils.getChildElementsByTagName(storedProcComponent, "sql-parameter-definition");
ManagedList<BeanDefinition> sqlParameterList = new ManagedList<BeanDefinition>();
for (Element childElement : sqlParameterDefinitionChildElements) {
String name = childElement.getAttribute("name");
String name = childElement.getAttribute(AbstractBeanDefinitionParser.NAME_ATTRIBUTE);
String sqlType = childElement.getAttribute("type");
String direction = childElement.getAttribute("direction");
String scale = childElement.getAttribute("scale");
@@ -163,13 +165,13 @@ public final class StoredProcParserUtils {
BeanDefinitionBuilder parameterBuilder = BeanDefinitionBuilder.genericBeanDefinition(ProcedureParameter.class);
String name = childElement.getAttribute("name");
String name = childElement.getAttribute(AbstractBeanDefinitionParser.NAME_ATTRIBUTE);
String expression = childElement.getAttribute("expression");
String value = childElement.getAttribute("value");
String type = childElement.getAttribute("type");
if (StringUtils.hasText(name)) {
parameterBuilder.addPropertyValue("name", name);
parameterBuilder.addPropertyValue(AbstractBeanDefinitionParser.NAME_ATTRIBUTE, name);
}
if (StringUtils.hasText(expression)) {
@@ -219,7 +221,7 @@ public final class StoredProcParserUtils {
for (Element childElement : returningResultsetChildElements) {
String name = childElement.getAttribute("name");
String name = childElement.getAttribute(AbstractBeanDefinitionParser.NAME_ATTRIBUTE);
String rowMapperAsString = childElement.getAttribute("row-mapper");
BeanMetadataElement rowMapperBeanDefinition = null;
@@ -229,7 +231,7 @@ public final class StoredProcParserUtils {
ClassUtils.forName(rowMapperAsString, parserContext.getReaderContext().getBeanClassLoader());
rowMapperBeanDefinition = BeanDefinitionBuilder.genericBeanDefinition(rowMapperAsString).getBeanDefinition();
}
catch (ClassNotFoundException e) {
catch (@SuppressWarnings("unused") ClassNotFoundException e) {
//Ignore it and fallback to bean reference
rowMapperBeanDefinition = new RuntimeBeanReference(rowMapperAsString);
}

12
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/metadata/JdbcMetadataStore.java
View File

@@ -42,6 +42,8 @@ import org.springframework.util.Assert;
*/
public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingBean {
private static final String KEY_CANNOT_BE_NULL = "'key' cannot be null";
/**
* Default value for the table prefix property.
*/
@@ -134,7 +136,7 @@ public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingB
@Override
@Transactional
public String putIfAbsent(String key, String value) {
Assert.notNull(key, "'key' cannot be null");
Assert.notNull(key, KEY_CANNOT_BE_NULL);
Assert.notNull(value, "'value' cannot be null");
while (true) {
//try to insert if does not exists
@@ -169,7 +171,7 @@ public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingB
@Override
@Transactional
public boolean replace(String key, String oldValue, String newValue) {
Assert.notNull(key, "'key' cannot be null");
Assert.notNull(key, KEY_CANNOT_BE_NULL);
Assert.notNull(oldValue, "'oldValue' cannot be null");
Assert.notNull(newValue, "'newValue' cannot be null");
int affectedRows = this.jdbcTemplate.update(this.replaceValueQuery,
@@ -185,7 +187,7 @@ public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingB
@Override
@Transactional
public void put(String key, String value) {
Assert.notNull(key, "'key' cannot be null");
Assert.notNull(key, KEY_CANNOT_BE_NULL);
Assert.notNull(value, "'value' cannot be null");
while (true) {
//try to insert if does not exist, if exists we will try to update it
@@ -215,7 +217,7 @@ public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingB
@Override
@Transactional
public String get(String key) {
Assert.notNull(key, "'key' cannot be null");
Assert.notNull(key, KEY_CANNOT_BE_NULL);
try {
return this.jdbcTemplate.queryForObject(this.getValueQuery, String.class, key, this.region);
}
@@ -228,7 +230,7 @@ public class JdbcMetadataStore implements ConcurrentMetadataStore, InitializingB
@Override
@Transactional
public String remove(String key) {
Assert.notNull(key, "'key' cannot be null");
Assert.notNull(key, KEY_CANNOT_BE_NULL);
String oldValue;
try {
//select old value and lock row for removal

29
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/DerbyChannelMessageStoreQueryProvider.java
View File

@@ -19,39 +19,42 @@ package org.springframework.integration.jdbc.store.channel;
/**
* @author Gunnar Hillert
* @author Artem Bilan
* @author Gary Russell
* @since 2.2
*
* https://blogs.oracle.com/kah/entry/derby_10_5_preview_fetch
*/
public class DerbyChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider {
private static final String SELECT_COMMON =
"SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES "
+ "from %PREFIX%CHANNEL_MESSAGE "
+ "where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region ";
@Override
public String getPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE, MESSAGE_SEQUENCE FETCH FIRST ROW ONLY";
return SELECT_COMMON
+ "and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) "
+ "order by CREATED_DATE, MESSAGE_SEQUENCE FETCH FIRST ROW ONLY";
}
@Override
public String getPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"order by CREATED_DATE, MESSAGE_SEQUENCE FETCH FIRST ROW ONLY";
return SELECT_COMMON
+ "order by CREATED_DATE, MESSAGE_SEQUENCE FETCH FIRST ROW ONLY";
}
@Override
public String getPriorityPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) " +
"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE FETCH FIRST ROW ONLY";
return SELECT_COMMON
+ "and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) "
+ "order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE FETCH FIRST ROW ONLY";
}
@Override
public String getPriorityPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE FETCH FIRST ROW ONLY";
return SELECT_COMMON
+ "order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE FETCH FIRST ROW ONLY";
}
}

22
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/H2ChannelMessageStoreQueryProvider.java
View File

@@ -20,11 +20,17 @@ package org.springframework.integration.jdbc.store.channel;
* @author Gunnar Hillert
* @author Artem Bilan
* @author Manuel Jordan
* @author Gary Russell
* @since 4.3
*
*/
public class H2ChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider {
private static final String SELECT_COMMON =
"SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES "
+ "from %PREFIX%CHANNEL_MESSAGE "
+ "where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region ";
@Override
public String getCreateMessageQuery() {
return "INSERT into %PREFIX%CHANNEL_MESSAGE(MESSAGE_ID, GROUP_KEY, REGION, CREATED_DATE, MESSAGE_PRIORITY, " +
@@ -34,35 +40,27 @@ public class H2ChannelMessageStoreQueryProvider extends AbstractChannelMessageSt
@Override
public String getPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES " +
"from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
return SELECT_COMMON +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) " +
"order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}
@Override
public String getPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES " +
"from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
return SELECT_COMMON +
"order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}
@Override
public String getPriorityPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES " +
"from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
return SELECT_COMMON +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) " +
"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}
@Override
public String getPriorityPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES " +
"from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
return SELECT_COMMON +
"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}

17
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/HsqlChannelMessageStoreQueryProvider.java
View File

@@ -24,6 +24,11 @@ package org.springframework.integration.jdbc.store.channel;
*/
public class HsqlChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider {
private static final String SELECT_COMMON =
"SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES "
+ "from %PREFIX%CHANNEL_MESSAGE "
+ "where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region ";
@Override
public String getCreateMessageQuery() {
return "INSERT into %PREFIX%CHANNEL_MESSAGE(MESSAGE_ID, GROUP_KEY, REGION, CREATED_DATE, MESSAGE_PRIORITY, MESSAGE_SEQUENCE, MESSAGE_BYTES)"
@@ -32,30 +37,26 @@ public class HsqlChannelMessageStoreQueryProvider extends AbstractChannelMessage
@Override
public String getPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
return SELECT_COMMON +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}
@Override
public String getPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
return SELECT_COMMON +
"order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}
@Override
public String getPriorityPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
return SELECT_COMMON +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) " +
"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}
@Override
public String getPriorityPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
return SELECT_COMMON +
"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}

20
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/MySqlChannelMessageStoreQueryProvider.java
View File

@@ -23,32 +23,34 @@ package org.springframework.integration.jdbc.store.channel;
*/
public class MySqlChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider {
private static final String SELECT_COMMON =
"SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES "
+ "from %PREFIX%CHANNEL_MESSAGE "
+ "where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region ";
@Override
public String getPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
return SELECT_COMMON
+ "and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) "
+ "order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}
@Override
public String getPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
return SELECT_COMMON +
"order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}
@Override
public String getPriorityPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
return SELECT_COMMON +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) " +
"order by MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}
@Override
public String getPriorityPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
return SELECT_COMMON +
"order by MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1";
}

17
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/OracleChannelMessageStoreQueryProvider.java
View File

@@ -33,23 +33,28 @@ package org.springframework.integration.jdbc.store.channel;
*/
public class OracleChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider {
private static final String SELECT_COMMON =
"SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES "
+ "from %PREFIX%CHANNEL_MESSAGE "
+ "where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region ";
@Override
public String getCreateMessageQuery() {
return "INSERT into %PREFIX%CHANNEL_MESSAGE(MESSAGE_ID, GROUP_KEY, REGION, CREATED_DATE, MESSAGE_PRIORITY, MESSAGE_SEQUENCE, MESSAGE_BYTES)"
return "INSERT into %PREFIX%CHANNEL_MESSAGE(MESSAGE_ID, GROUP_KEY, REGION, CREATED_DATE, MESSAGE_PRIORITY, "
+ "MESSAGE_SEQUENCE, MESSAGE_BYTES)"
+ " values (?, ?, ?, ?, ?, %PREFIX%MESSAGE_SEQ.NEXTVAL, ?)";
}
@Override
public String getPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE, MESSAGE_SEQUENCE FOR UPDATE SKIP LOCKED";
return SELECT_COMMON
+ "and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) "
+ "order by CREATED_DATE, MESSAGE_SEQUENCE FOR UPDATE SKIP LOCKED";
}
@Override
public String getPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
return SELECT_COMMON +
"order by CREATED_DATE, MESSAGE_SEQUENCE FOR UPDATE SKIP LOCKED";
}

20
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/PostgresChannelMessageStoreQueryProvider.java
View File

@@ -23,32 +23,34 @@ package org.springframework.integration.jdbc.store.channel;
*/
public class PostgresChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider {
private static final String SELECT_COMMON =
"SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES "
+ "from %PREFIX%CHANNEL_MESSAGE "
+ "where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region ";
@Override
public String getPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE";
return SELECT_COMMON
+ "and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) "
+ "order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE";
}
@Override
public String getPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
return SELECT_COMMON +
"order by CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE";
}
@Override
public String getPriorityPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
return SELECT_COMMON +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) " +
"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE";
}
@Override
public String getPriorityPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
return SELECT_COMMON +
"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE LIMIT 1 FOR UPDATE";
}

20
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/SqlServerChannelMessageStoreQueryProvider.java
View File

@@ -23,38 +23,40 @@ package org.springframework.integration.jdbc.store.channel;
*/
public class SqlServerChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider {
private static final String SELECT_COMMON =
"SELECT TOP 1 %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES "
+ "from %PREFIX%CHANNEL_MESSAGE "
+ "where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region ";
@Override
public String getPollFromGroupExcludeIdsQuery() {
return "SELECT TOP 1 %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
return SELECT_COMMON +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) order by CREATED_DATE, MESSAGE_SEQUENCE";
}
@Override
public String getPollFromGroupQuery() {
return "SELECT TOP 1 %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
return SELECT_COMMON +
"order by CREATED_DATE, MESSAGE_SEQUENCE";
}
@Override
public String getPriorityPollFromGroupExcludeIdsQuery() {
return "SELECT TOP 1 %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
return SELECT_COMMON +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) " +
"order by MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE";
}
@Override
public String getPriorityPollFromGroupQuery() {
return "SELECT TOP 1 %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
return SELECT_COMMON +
"order by MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE";
}
@Override
public String getCreateMessageQuery() {
return "INSERT into %PREFIX%CHANNEL_MESSAGE(MESSAGE_ID, GROUP_KEY, REGION, CREATED_DATE, MESSAGE_PRIORITY, MESSAGE_SEQUENCE, MESSAGE_BYTES)"
return "INSERT into %PREFIX%CHANNEL_MESSAGE(MESSAGE_ID, GROUP_KEY, REGION, CREATED_DATE, MESSAGE_PRIORITY, "
+ "MESSAGE_SEQUENCE, MESSAGE_BYTES)"
+ " values (?, ?, ?, ?, ?,(NEXT VALUE FOR %PREFIX%MESSAGE_SEQ), ?)";
}

12
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/metadata/MongoDbMetadataStore.java
View File

@@ -44,6 +44,8 @@ import org.springframework.util.Assert;
*/
public class MongoDbMetadataStore implements ConcurrentMetadataStore {
private static final String KEY_MUST_NOT_BE_EMPTY = "'key' must not be empty.";
private static final String DEFAULT_COLLECTION_NAME = "metadataStore";
private static final String ID_FIELD = "_id";
@@ -107,7 +109,7 @@ public class MongoDbMetadataStore implements ConcurrentMetadataStore {
*/
@Override
public void put(String key, String value) {
Assert.hasText(key, "'key' must not be empty.");
Assert.hasText(key, KEY_MUST_NOT_BE_EMPTY);
Assert.hasText(value, "'value' must not be empty.");
final Map<String, Object> entry = new HashMap<>();
entry.put(ID_FIELD, key);
@@ -123,7 +125,7 @@ public class MongoDbMetadataStore implements ConcurrentMetadataStore {
*/
@Override
public String get(String key) {
Assert.hasText(key, "'key' must not be empty.");
Assert.hasText(key, KEY_MUST_NOT_BE_EMPTY);
Query query = new Query(Criteria.where(ID_FIELD).is(key));
query.fields().exclude(ID_FIELD);
@SuppressWarnings("unchecked")
@@ -141,7 +143,7 @@ public class MongoDbMetadataStore implements ConcurrentMetadataStore {
*/
@Override
public String remove(String key) {
Assert.hasText(key, "'key' must not be empty.");
Assert.hasText(key, KEY_MUST_NOT_BE_EMPTY);
Query query = new Query(Criteria.where(ID_FIELD).is(key));
query.fields().exclude(ID_FIELD);
@SuppressWarnings("unchecked")
@@ -167,7 +169,7 @@ public class MongoDbMetadataStore implements ConcurrentMetadataStore {
*/
@Override
public String putIfAbsent(String key, String value) {
Assert.hasText(key, "'key' must not be empty.");
Assert.hasText(key, KEY_MUST_NOT_BE_EMPTY);
Assert.hasText(value, "'value' must not be empty.");
Query query = new Query(Criteria.where(ID_FIELD).is(key));
@@ -190,7 +192,7 @@ public class MongoDbMetadataStore implements ConcurrentMetadataStore {
*/
@Override
public boolean replace(String key, String oldValue, String newValue) {
Assert.hasText(key, "'key' must not be empty.");
Assert.hasText(key, KEY_MUST_NOT_BE_EMPTY);
Assert.hasText(oldValue, "'oldValue' must not be empty.");
Assert.hasText(newValue, "'newValue' must not be empty.");
Query query = new Query(Criteria.where(ID_FIELD).is(key).and(VALUE).is(oldValue));

14
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java
View File

@@ -55,6 +55,8 @@ import org.springframework.util.Assert;
public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDbMessageStore
implements MessageStore {
private static final String GROUP_ID_MUST_NOT_BE_NULL = "'groupId' must not be null";
public static final String DEFAULT_COLLECTION_NAME = "configurableStoreMessages";
@@ -109,7 +111,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb
@Override
public MessageGroup getMessageGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL);
Query query = groupOrderQuery(groupId);
MessageDocument messageDocument = getMongoTemplate().findOne(query, MessageDocument.class, this.collectionName);
@@ -140,7 +142,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb
@Override
public void addMessagesToGroup(Object groupId, Message<?>... messages) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL);
Assert.notNull(messages, "'message' must not be null");
Query query = groupOrderQuery(groupId);
@@ -171,7 +173,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb
@Override
public void removeMessagesFromGroup(Object groupId, Collection<Message<?>> messages) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL);
Assert.notNull(messages, "'messageToRemove' must not be null");
Collection<UUID> ids = new ArrayList<>();
@@ -196,7 +198,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb
@Override
public Message<?> pollMessageFromGroup(final Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL);
Sort sort = Sort.by(MessageDocumentFields.LAST_MODIFIED_TIME, MessageDocumentFields.SEQUENCE);
Query query = groupIdQuery(groupId).with(sort);
@@ -253,7 +255,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb
@Override
public Message<?> getOneMessageFromGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL);
Query query = groupOrderQuery(groupId);
MessageDocument messageDocument = getMongoTemplate().findOne(query, MessageDocument.class, this.collectionName);
if (messageDocument != null) {
@@ -266,7 +268,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb
@Override
public Collection<Message<?>> getMessagesForGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL);
Query query = groupOrderQuery(groupId);
List<MessageDocument> documents = getMongoTemplate().find(query, MessageDocument.class, this.collectionName);

66
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java
View File

@@ -101,6 +101,12 @@ import com.mongodb.DBObject;
public class MongoDbMessageStore extends AbstractMessageGroupStore
implements MessageStore, BeanClassLoaderAware, ApplicationContextAware, InitializingBean {
private static final String HEADERS = "headers";
private static final String UNCHECKED = "unchecked";
private static final String GROUP_ID_MUST_NOT_BE_NULL = "'groupId' must not be null";
public static final String SEQUENCE_NAME = "messagesSequence";
private static final String DEFAULT_COLLECTION_NAME = "messages";
@@ -249,7 +255,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
@Override
public MessageGroup getMessageGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL);
Query query = whereGroupIdOrder(groupId);
MessageWrapper messageWrapper = this.template.findOne(query, MessageWrapper.class, this.collectionName);
@@ -273,7 +279,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
@Override
public void addMessagesToGroup(Object groupId, Message<?>... messages) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL);
Assert.notNull(messages, "'message' must not be null");
Query query = whereGroupIdOrder(groupId);
MessageWrapper messageDocument = this.template.findOne(query, MessageWrapper.class, this.collectionName);
@@ -303,7 +309,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
@Override
public void removeMessagesFromGroup(Object groupId, Collection<Message<?>> messages) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL);
Assert.notNull(messages, "'messageToRemove' must not be null");
Collection<UUID> ids = new ArrayList<>();
@@ -352,7 +358,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
@Override
public Message<?> pollMessageFromGroup(final Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL);
Query query = whereGroupIdIs(groupId).with(Sort.by(GROUP_UPDATE_TIMESTAMP_KEY, SEQUENCE));
MessageWrapper messageWrapper = this.template.findAndRemove(query, MessageWrapper.class, this.collectionName);
Message<?> message = null;
@@ -382,7 +388,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
@Override
public Message<?> getOneMessageFromGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL);
Query query = whereGroupIdOrder(groupId);
MessageWrapper messageWrapper = this.template.findOne(query, MessageWrapper.class, this.collectionName);
if (messageWrapper != null) {
@@ -395,7 +401,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
@Override
public Collection<Message<?>> getMessagesForGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(groupId, GROUP_ID_MUST_NOT_BE_NULL);
Query query = whereGroupIdOrder(groupId);
List<MessageWrapper> messageWrappers = this.template.find(query, MessageWrapper.class, this.collectionName);
@@ -463,10 +469,10 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
this.collectionName).get(SEQUENCE); // NOSONAR - never returns null
}
@SuppressWarnings("unchecked")
@SuppressWarnings(UNCHECKED)
private static void enhanceHeaders(MessageHeaders messageHeaders, Map<String, Object> headers) {
Map<String, Object> innerMap =
(Map<String, Object>) new DirectFieldAccessor(messageHeaders).getPropertyValue("headers");
(Map<String, Object>) new DirectFieldAccessor(messageHeaders).getPropertyValue(HEADERS);
// using reflection to set ID and TIMESTAMP since they are immutable through MessageHeaders
Object idHeader = headers.get(MessageHeaders.ID);
if (idHeader != null) {
@@ -478,7 +484,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
}
}
@SuppressWarnings("unchecked")
@SuppressWarnings(UNCHECKED)
private static Map<String, Object> asMap(Bson bson) {
if (bson instanceof Document) {
return (Document) bson;
@@ -498,6 +504,8 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
*/
private final class MessageReadingMongoConverter extends MappingMongoConverter {
private static final String CLASS = "_class";
MessageReadingMongoConverter(MongoDbFactory mongoDbFactory,
MappingContext<? extends MongoPersistentEntity<?>, MongoPersistentProperty> mappingContext) {
super(new DefaultDbRefResolver(mongoDbFactory), mappingContext);
@@ -531,7 +539,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
}
@Override
@SuppressWarnings({ "unchecked" })
@SuppressWarnings({ UNCHECKED })
public <S> S read(Class<S> clazz, Bson source) {
if (!MessageWrapper.class.equals(clazz)) {
return super.read(clazz, source);
@@ -590,8 +598,8 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
Map<String, Object> document = asMap(source);
try {
Class<?> typeClass = null;
if (document.containsKey("_class")) {
Object type = document.get("_class");
if (document.containsKey(CLASS)) {
Object type = document.get(CLASS);
typeClass = ClassUtils.forName(type.toString(), MongoDbMessageStore.this.classLoader);
}
else if (source instanceof BasicDBList) {
@@ -618,7 +626,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
if (payload instanceof Bson) {
Bson payloadObject = (Bson) payload;
Object payloadType = asMap(payloadObject).get("_class");
Object payloadType = asMap(payloadObject).get(CLASS);
try {
Class<?> payloadClass =
ClassUtils.forName(payloadType.toString(), MongoDbMessageStore.this.classLoader);
@@ -667,9 +675,9 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
@Override
public GenericMessage<?> convert(Document source) {
@SuppressWarnings("unchecked")
@SuppressWarnings(UNCHECKED)
Map<String, Object> headers =
MongoDbMessageStore.this.converter.normalizeHeaders((Map<String, Object>) source.get("headers"));
MongoDbMessageStore.this.converter.normalizeHeaders((Map<String, Object>) source.get(HEADERS));
GenericMessage<?> message =
new GenericMessage<>(MongoDbMessageStore.this.converter.extractPayload(source), headers);
@@ -688,9 +696,9 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
@Override
public MutableMessage<?> convert(Document source) {
@SuppressWarnings("unchecked")
@SuppressWarnings(UNCHECKED)
Map<String, Object> headers =
MongoDbMessageStore.this.converter.normalizeHeaders((Map<String, Object>) source.get("headers"));
MongoDbMessageStore.this.converter.normalizeHeaders((Map<String, Object>) source.get(HEADERS));
Object payload = MongoDbMessageStore.this.converter.extractPayload(source);
return (MutableMessage<?>) MutableMessageBuilder.withPayload(payload)
@@ -709,9 +717,9 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
@Override
public AdviceMessage<?> convert(Document source) {
@SuppressWarnings("unchecked")
@SuppressWarnings(UNCHECKED)
Map<String, Object> headers =
MongoDbMessageStore.this.converter.normalizeHeaders((Map<String, Object>) source.get("headers"));
MongoDbMessageStore.this.converter.normalizeHeaders((Map<String, Object>) source.get(HEADERS));
Message<?> inputMessage = null;
@@ -749,9 +757,9 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
@Override
public ErrorMessage convert(Document source) {
@SuppressWarnings("unchecked")
@SuppressWarnings(UNCHECKED)
Map<String, Object> headers =
MongoDbMessageStore.this.converter.normalizeHeaders((Map<String, Object>) source.get("headers"));
MongoDbMessageStore.this.converter.normalizeHeaders((Map<String, Object>) source.get(HEADERS));
Object payload = this.deserializingConverter.convert(((Binary) source.get("payload")).getData());
ErrorMessage message = new ErrorMessage((Throwable) payload, headers); // NOSONAR not null
@@ -784,12 +792,14 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
*/
private static final class MessageWrapper {
private static final String UNUSED = "unused";
/*
* Needed as a persistence property to suppress 'Cannot determine IsNewStrategy' MappingException
* when the application context is configured with auditing. The document is not
* currently Auditable.
*/
@SuppressWarnings("unused")
@SuppressWarnings(UNUSED)
@Id
private String _id; // NOSONAR name
@@ -798,16 +808,16 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
@Transient
private final Message<?> message; // NOSONAR name
@SuppressWarnings("unused")
@SuppressWarnings(UNUSED)
private final String _messageType; // NOSONAR name
@SuppressWarnings("unused")
@SuppressWarnings(UNUSED)
private final Object payload;
@SuppressWarnings("unused")
@SuppressWarnings(UNUSED)
private final Map<String, ?> headers;
@SuppressWarnings("unused")
@SuppressWarnings(UNUSED)
private final Message<?> inputMessage;
private long _message_timestamp; // NOSONAR name
@@ -820,7 +830,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
private volatile boolean _group_complete; // NOSONAR name
@SuppressWarnings("unused")
@SuppressWarnings(UNUSED)
private int sequence;
MessageWrapper(Message<?> message) {
@@ -849,7 +859,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
return this._group_complete;
}
@SuppressWarnings("unused")
@SuppressWarnings(UNUSED)
public Object get_GroupId() { // NOSONAR name
return this._groupId;
}

12
spring-integration-redis/src/main/java/org/springframework/integration/redis/metadata/RedisMetadataStore.java
View File

@@ -40,6 +40,8 @@ import org.springframework.util.Assert;
*/
public class RedisMetadataStore implements ConcurrentMetadataStore {
private static final String KEY_MUST_NOT_BE_NULL = "'key' must not be null.";
public static final String KEY = "MetaData";
private final RedisProperties properties;
@@ -109,7 +111,7 @@ public class RedisMetadataStore implements ConcurrentMetadataStore {
*/
@Override
public void put(String key, String value) {
Assert.notNull(key, "'key' must not be null.");
Assert.notNull(key, KEY_MUST_NOT_BE_NULL);
Assert.notNull(value, "'value' must not be null.");
this.properties.put(key, value);
}
@@ -121,7 +123,7 @@ public class RedisMetadataStore implements ConcurrentMetadataStore {
*/
@Override
public String get(String key) {
Assert.notNull(key, "'key' must not be null.");
Assert.notNull(key, KEY_MUST_NOT_BE_NULL);
Object value = this.properties.get(key);
if (value != null) {
Assert.isInstanceOf(String.class, value, "Invalid type in the store");
@@ -132,7 +134,7 @@ public class RedisMetadataStore implements ConcurrentMetadataStore {
@Override
public String remove(String key) {
Assert.notNull(key, "'key' must not be null.");
Assert.notNull(key, KEY_MUST_NOT_BE_NULL);
Object removed = this.properties.remove(key);
if (removed != null) {
Assert.isInstanceOf(String.class, removed, "The removed value was an invalid type");
@@ -142,7 +144,7 @@ public class RedisMetadataStore implements ConcurrentMetadataStore {
@Override
public String putIfAbsent(String key, String value) {
Assert.notNull(key, "'key' must not be null.");
Assert.notNull(key, KEY_MUST_NOT_BE_NULL);
Assert.notNull(value, "'value' must not be null.");
Object oldValue = this.properties.putIfAbsent(key, value);
if (oldValue != null) {
@@ -153,7 +155,7 @@ public class RedisMetadataStore implements ConcurrentMetadataStore {
@Override
public boolean replace(String key, String oldValue, String newValue) {
Assert.notNull(key, "'key' must not be null.");
Assert.notNull(key, KEY_MUST_NOT_BE_NULL);
Assert.notNull(oldValue, "'oldValue' must not be null.");
Assert.notNull(newValue, "'newValue' must not be null.");
return this.properties.replace(key, oldValue, newValue);

10
spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java
View File

@@ -43,6 +43,8 @@ import org.springframework.util.Assert;
*/
public class RedisMessageStore extends AbstractKeyValueMessageStore implements BeanClassLoaderAware {
private static final String ID_MUST_NOT_BE_NULL = "'id' must not be null";
private final RedisTemplate<Object, Object> redisTemplate;
private final boolean unlinkAvailable;
@@ -92,7 +94,7 @@ public class RedisMessageStore extends AbstractKeyValueMessageStore implements B
@Override
protected Object doRetrieve(Object id) {
Assert.notNull(id, "'id' must not be null");
Assert.notNull(id, ID_MUST_NOT_BE_NULL);
BoundValueOperations<Object, Object> ops = this.redisTemplate.boundValueOps(id);
return ops.get();
}
@@ -100,7 +102,7 @@ public class RedisMessageStore extends AbstractKeyValueMessageStore implements B
@Override
protected void doStore(Object id, Object objectToStore) {
Assert.notNull(id, "'id' must not be null");
Assert.notNull(id, ID_MUST_NOT_BE_NULL);
Assert.notNull(objectToStore, "'objectToStore' must not be null");
BoundValueOperations<Object, Object> ops = this.redisTemplate.boundValueOps(id);
try {
@@ -114,7 +116,7 @@ public class RedisMessageStore extends AbstractKeyValueMessageStore implements B
@Override
protected void doStoreIfAbsent(Object id, Object objectToStore) {
Assert.notNull(id, "'id' must not be null");
Assert.notNull(id, ID_MUST_NOT_BE_NULL);
Assert.notNull(objectToStore, "'objectToStore' must not be null");
BoundValueOperations<Object, Object> ops = this.redisTemplate.boundValueOps(id);
try {
@@ -131,7 +133,7 @@ public class RedisMessageStore extends AbstractKeyValueMessageStore implements B
@Override
protected Object doRemove(Object id) {
Assert.notNull(id, "'id' must not be null");
Assert.notNull(id, ID_MUST_NOT_BE_NULL);
Object removedObject = this.doRetrieve(id);
if (removedObject != null) {
if (this.unlinkAvailable) {

12
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java
View File

@@ -49,6 +49,8 @@ import com.jcraft.jsch.SftpException;
*/
public class SftpSession implements Session<LsEntry> {
private static final String SESSION_IS_NOT_CONNECTED = "session is not connected";
private final Log logger = LogFactory.getLog(this.getClass());
private final com.jcraft.jsch.Session jschSession;
@@ -74,7 +76,7 @@ public class SftpSession implements Session<LsEntry> {
@Override
public boolean remove(String path) throws IOException {
Assert.state(this.channel != null, "session is not connected");
Assert.state(this.channel != null, SESSION_IS_NOT_CONNECTED);
try {
this.channel.rm(path);
return true;
@@ -86,7 +88,7 @@ public class SftpSession implements Session<LsEntry> {
@Override
public LsEntry[] list(String path) throws IOException {
Assert.state(this.channel != null, "session is not connected");
Assert.state(this.channel != null, SESSION_IS_NOT_CONNECTED);
try {
Vector<?> lsEntries = this.channel.ls(path); // NOSONAR (Vector)
if (lsEntries != null) {
@@ -123,7 +125,7 @@ public class SftpSession implements Session<LsEntry> {
@Override
public void read(String source, OutputStream os) throws IOException {
Assert.state(this.channel != null, "session is not connected");
Assert.state(this.channel != null, SESSION_IS_NOT_CONNECTED);
try {
InputStream is = this.channel.get(source);
FileCopyUtils.copy(is, os);
@@ -150,7 +152,7 @@ public class SftpSession implements Session<LsEntry> {
@Override
public void write(InputStream inputStream, String destination) throws IOException {
Assert.state(this.channel != null, "session is not connected");
Assert.state(this.channel != null, SESSION_IS_NOT_CONNECTED);
try {
this.channel.put(inputStream, destination);
}
@@ -161,7 +163,7 @@ public class SftpSession implements Session<LsEntry> {
@Override
public void append(InputStream inputStream, String destination) throws IOException {
Assert.state(this.channel != null, "session is not connected");
Assert.state(this.channel != null, SESSION_IS_NOT_CONNECTED);
try {
this.channel.put(inputStream, destination, ChannelSftp.APPEND);
}

24
spring-integration-test-support/src/main/java/org/springframework/integration/test/mail/TestMailServer.java
View File

@@ -169,6 +169,8 @@ public final class TestMailServer {
class Pop3Handler extends MailHandler {
private static final String PLUS_OK = "+OK";
Pop3Handler(Socket socket) {
super(socket);
}
@@ -180,29 +182,29 @@ public final class TestMailServer {
while (!socket.isClosed()) {
String line = reader.readLine();
if ("CAPA".equals(line)) {
write("+OK");
write(PLUS_OK);
write("USER");
write(".");
}
else if ("USER user".equals(line)) {
write("+OK");
write(PLUS_OK);
}
else if ("PASS pw".equals(line)) {
write("+OK");
write(PLUS_OK);
}
else if ("STAT".equals(line)) {
write("+OK 1 3");
}
else if ("NOOP".equals(line)) {
write("+OK");
write(PLUS_OK);
}
else if ("RETR 1".equals(line)) {
write("+OK");
write(PLUS_OK);
write(MESSAGE);
write(".");
}
else if ("QUIT".equals(line)) {
write("+OK");
write(PLUS_OK);
socket.close();
}
}
@@ -240,6 +242,8 @@ public final class TestMailServer {
class ImapHandler extends MailHandler {
private static final String OK_FETCH_COMPLETED = "OK FETCH completed";
/**
* Time to wait while IDLE before returning a result.
*/
@@ -314,13 +318,13 @@ public final class TestMailServer {
+ "\"<CACVnpJkAUUfa3d_-4GNZW2WpxbB39tBCHC=T0gc7hty6dOEHcA@foo.bar.com>\") " // msgid
+ "BODYSTRUCTURE "
+ "(\"TEXT\" \"PLAIN\" (\"CHARSET\" \"ISO-8859-1\") NIL NIL \"7BIT\" 1 5)))");
write(tag + "OK FETCH completed");
write(tag + OK_FETCH_COMPLETED);
}
else if (line.contains("FETCH 2 (BODYSTRUCTURE)")) {
write("* 2 FETCH " +
"BODYSTRUCTURE "
+ "(\"TEXT\" \"PLAIN\" (\"CHARSET\" \"ISO-8859-1\") NIL NIL \"7BIT\" 1 5)))");
write(tag + "OK FETCH completed");
write(tag + OK_FETCH_COMPLETED);
}
else if (line.contains("STORE 1 +FLAGS (\\Flagged)")) {
write("* 1 FETCH (FLAGS (\\Flagged))");
@@ -333,13 +337,13 @@ public final class TestMailServer {
}
else if (line.contains("FETCH 1 FLAGS")) {
write("* 1 FLAGS(\\Seen)");
write(tag + "OK FETCH completed");
write(tag + OK_FETCH_COMPLETED);
}
else if (line.contains("FETCH 1 (BODY.PEEK")) {
write("* 1 FETCH (BODY[]<0> {" + (MESSAGE.length() + 2) + "}");
write(MESSAGE);
write(")");
write(tag + "OK FETCH completed");
write(tag + OK_FETCH_COMPLETED);
}
else if (line.contains("CLOSE")) {
write(tag + "OK CLOSE completed");

13
spring-integration-test/src/main/java/org/springframework/integration/test/context/MockIntegrationContext.java
View File

@@ -58,6 +58,11 @@ import reactor.util.function.Tuples;
*/
public class MockIntegrationContext implements BeanFactoryAware {
private static final String HANDLER = "handler";
/**
* The bean name for the mock integration context.
*/
public static final String MOCK_INTEGRATION_CONTEXT_BEAN_NAME = "mockIntegrationContext";
private final Map<String, Object> beans = new HashMap<>();
@@ -97,11 +102,11 @@ public class MockIntegrationContext implements BeanFactoryAware {
}
else if (endpoint instanceof ReactiveStreamsConsumer) {
Tuple2<?, ?> value = (Tuple2<?, ?>) e.getValue();
directFieldAccessor.setPropertyValue("handler", value.getT1());
directFieldAccessor.setPropertyValue(HANDLER, value.getT1());
directFieldAccessor.setPropertyValue("subscriber", value.getT2());
}
else if (endpoint instanceof IntegrationConsumer) {
directFieldAccessor.setPropertyValue("handler", e.getValue());
directFieldAccessor.setPropertyValue(HANDLER, e.getValue());
}
});
@@ -147,7 +152,7 @@ public class MockIntegrationContext implements BeanFactoryAware {
((Lifecycle) endpoint).stop();
}
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(endpoint);
Object targetMessageHandler = directFieldAccessor.getPropertyValue("handler");
Object targetMessageHandler = directFieldAccessor.getPropertyValue(HANDLER);
Assert.notNull(targetMessageHandler, () -> "'handler' must not be null in the: " + endpoint);
if (endpoint instanceof ReactiveStreamsConsumer) {
Object targetSubscriber = directFieldAccessor.getPropertyValue("subscriber");
@@ -177,7 +182,7 @@ public class MockIntegrationContext implements BeanFactoryAware {
}
}
directFieldAccessor.setPropertyValue("handler", mockMessageHandler);
directFieldAccessor.setPropertyValue(HANDLER, mockMessageHandler);
if (endpoint instanceof ReactiveStreamsConsumer) {
directFieldAccessor.setPropertyValue("subscriber", mockMessageHandler);

10
spring-integration-xml/src/main/java/org/springframework/integration/xml/selector/RegexTestXPathMessageSelector.java
View File

@@ -35,6 +35,8 @@ import org.springframework.xml.xpath.XPathExpression;
*/
public class RegexTestXPathMessageSelector extends AbstractXPathMessageSelector {
private static final String REGEX_MUST_NOT_BE_NULL = "regex must not be null";
private final String regex;
@@ -47,7 +49,7 @@ public class RegexTestXPathMessageSelector extends AbstractXPathMessageSelector
*/
public RegexTestXPathMessageSelector(String expression, Map<String, String> namespaces, String regex) {
super(expression, namespaces);
Assert.notNull(regex, "regex must not be null");
Assert.notNull(regex, REGEX_MUST_NOT_BE_NULL);
this.regex = regex;
}
@@ -61,7 +63,7 @@ public class RegexTestXPathMessageSelector extends AbstractXPathMessageSelector
*/
public RegexTestXPathMessageSelector(String expression, String prefix, String namespace, String regex) {
super(expression, prefix, namespace);
Assert.notNull(regex, "regex must not be null");
Assert.notNull(regex, REGEX_MUST_NOT_BE_NULL);
this.regex = regex;
}
@@ -73,7 +75,7 @@ public class RegexTestXPathMessageSelector extends AbstractXPathMessageSelector
*/
public RegexTestXPathMessageSelector(String expression, String regex) {
super(expression);
Assert.notNull(regex, "regex must not be null");
Assert.notNull(regex, REGEX_MUST_NOT_BE_NULL);
this.regex = regex;
}
@@ -86,7 +88,7 @@ public class RegexTestXPathMessageSelector extends AbstractXPathMessageSelector
*/
public RegexTestXPathMessageSelector(XPathExpression expression, String regex) {
super(expression);
Assert.notNull(regex, "regex must not be null");
Assert.notNull(regex, REGEX_MUST_NOT_BE_NULL);
this.regex = regex;
}

12
spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStore.java
View File

@@ -48,6 +48,8 @@ import org.springframework.util.Assert;
*/
public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLifecycle {
private static final String KEY_MUST_NOT_BE_NULL = "'key' must not be null.";
private static final String UNUSED = "unused";
private final Object lifecycleMonitor = new Object();
@@ -113,7 +115,7 @@ public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLif
@Override
public String putIfAbsent(String key, String value) {
Assert.notNull(key, "'key' must not be null.");
Assert.notNull(key, KEY_MUST_NOT_BE_NULL);
Assert.notNull(value, "'value' must not be null.");
synchronized (this.updateMap) {
try {
@@ -132,7 +134,7 @@ public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLif
@Override
public boolean replace(String key, String oldValue, String newValue) {
Assert.notNull(key, "'key' must not be null.");
Assert.notNull(key, KEY_MUST_NOT_BE_NULL);
Assert.notNull(oldValue, "'oldValue' must not be null.");
Assert.notNull(newValue, "'newValue' must not be null.");
synchronized (this.updateMap) {
@@ -171,7 +173,7 @@ public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLif
@Override
public void put(String key, String value) {
Assert.notNull(key, "'key' must not be null.");
Assert.notNull(key, KEY_MUST_NOT_BE_NULL);
Assert.notNull(value, "'value' must not be null.");
synchronized (this.updateMap) {
try {
@@ -196,7 +198,7 @@ public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLif
@Override
public String get(String key) {
Assert.notNull(key, "'key' must not be null.");
Assert.notNull(key, KEY_MUST_NOT_BE_NULL);
Assert.state(isRunning(), "ZookeeperMetadataStore has to be started before using.");
synchronized (this.updateMap) {
ChildData currentData = this.cache.getCurrentData(getPath(key));
@@ -225,7 +227,7 @@ public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLif
@Override
public String remove(String key) {
Assert.notNull(key, "'key' must not be null.");
Assert.notNull(key, KEY_MUST_NOT_BE_NULL);
synchronized (this.updateMap) {
try {
byte[] bytes = this.client.getData().forPath(getPath(key));