diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java index 79155ca3de..75349a4035 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java @@ -385,21 +385,25 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement } } if (delayValueException != null) { - if (this.ignoreExpressionFailures) { - if (logger.isDebugEnabled()) { - logger.debug("Failed to get delay value from 'delayExpression': " + - delayValueException.getMessage() + - ". Will fall back to default delay: " + this.defaultDelay); - } - } - else { - throw new IllegalStateException("Error occurred during 'delay' value determination", - delayValueException); - } + handleDelayValueException(delayValueException); } return delay; } + private void handleDelayValueException(Exception delayValueException) { + if (this.ignoreExpressionFailures) { + if (logger.isDebugEnabled()) { + logger.debug("Failed to get delay value from 'delayExpression': " + + delayValueException.getMessage() + + ". Will fall back to default delay: " + this.defaultDelay); + } + } + else { + throw new IllegalStateException("Error occurred during 'delay' value determination", + delayValueException); + } + } + private void releaseMessageAfterDelay(final Message message, long delay) { Message delayedMessage = message; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMappingMessageRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMappingMessageRouter.java index 4d384e40b0..0655366669 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMappingMessageRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMappingMessageRouter.java @@ -27,6 +27,7 @@ import java.util.Map.Entry; import java.util.Properties; import java.util.Set; +import org.springframework.core.convert.ConversionService; import org.springframework.integration.support.management.MappingMessageRouterManagement; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.jmx.export.annotation.ManagedOperation; @@ -292,37 +293,41 @@ public abstract class AbstractMappingMessageRouter extends AbstractMessageRouter return; } for (Object channelKey : channelKeys) { - if (channelKey instanceof MessageChannel) { - channels.add((MessageChannel) channelKey); + addChannelKeyToCollection(channels, message, channelKey); + } + } + + private void addChannelKeyToCollection(Collection channels, Message message, Object channelKey) { + ConversionService conversionService = getRequiredConversionService(); + if (channelKey instanceof MessageChannel) { + channels.add((MessageChannel) channelKey); + } + else if (channelKey instanceof MessageChannel[]) { + channels.addAll(Arrays.asList((MessageChannel[]) channelKey)); + } + else if (channelKey instanceof String) { + addChannelFromString(channels, (String) channelKey, message); + } + else if (channelKey instanceof Class) { + addChannelFromString(channels, ((Class) channelKey).getName(), message); + } + else if (channelKey instanceof String[]) { + for (String indicatorName : (String[]) channelKey) { + addChannelFromString(channels, indicatorName, message); } - else if (channelKey instanceof MessageChannel[]) { - channels.addAll(Arrays.asList((MessageChannel[]) channelKey)); - } - else if (channelKey instanceof String) { - addChannelFromString(channels, (String) channelKey, message); - } - else if (channelKey instanceof Class) { - addChannelFromString(channels, ((Class) channelKey).getName(), message); - } - else if (channelKey instanceof String[]) { - for (String indicatorName : (String[]) channelKey) { - addChannelFromString(channels, indicatorName, message); - } - } - else if (channelKey instanceof Collection) { - addToCollection(channels, (Collection) channelKey, message); - } - else if (channelKey != null && - getRequiredConversionService().canConvert(channelKey.getClass(), String.class)) { - String converted = getConversionService().convert(channelKey, String.class); - if (converted != null) { - addChannelFromString(channels, converted, message); - } - } - else if (channelKey != null) { - throw new MessagingException("unsupported return type for router [" + channelKey.getClass() + "]"); + } + else if (channelKey instanceof Collection) { + addToCollection(channels, (Collection) channelKey, message); + } + else if (channelKey != null && conversionService.canConvert(channelKey.getClass(), String.class)) { + String converted = conversionService.convert(channelKey, String.class); + if (converted != null) { + addChannelFromString(channels, converted, message); } } + else if (channelKey != null) { + throw new MessagingException("unsupported return type for router [" + channelKey.getClass() + "]"); + } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java b/spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java index 6ff2db2ccf..4fb49b933d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -391,50 +391,9 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe } } catch (Exception e) { - if (this.locked) { - this.locked = false; - try { - this.lock.unlock(); - } - catch (Exception e1) { - logger.debug("Could not unlock - treat as broken " + this.context + - ". Revoking " + (isRunning() ? " and retrying..." : "..."), e1); - - } - // The lock was broken and we are no longer leader - handleRevoked(); - } - - if (e instanceof InterruptedException || Thread.currentThread().isInterrupted()) { - Thread.currentThread().interrupt(); - if (isRunning()) { - logger.warn("Restarting LeaderSelector for " + this.context + " because of error.", e); - LockRegistryLeaderInitiator.this.future = - LockRegistryLeaderInitiator.this.executorService.submit( - () -> { - // Give it a chance to elect some other leader. - Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis); - return call(); - }); - } + if (handleLockException(e)) { return null; } - else { - if (isRunning()) { - // Give it a chance to elect some other leader. - try { - Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis); - } - catch (InterruptedException e1) { - // Ignore interruption and let it to be caught on the next cycle. - Thread.currentThread().interrupt(); - } - } - if (logger.isDebugEnabled()) { - logger.debug("Error acquiring the lock for " + this.context + - ". " + (isRunning() ? "Retrying..." : ""), e); - } - } } } } @@ -455,6 +414,54 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe return null; } + private boolean handleLockException(Exception e) { + if (this.locked) { + this.locked = false; + try { + this.lock.unlock(); + } + catch (Exception e1) { + logger.debug("Could not unlock - treat as broken " + this.context + + ". Revoking " + (isRunning() ? " and retrying..." : "..."), e1); + + } + // The lock was broken and we are no longer leader + handleRevoked(); + } + + if (e instanceof InterruptedException || Thread.currentThread().isInterrupted()) { + Thread.currentThread().interrupt(); + if (isRunning()) { + logger.warn("Restarting LeaderSelector for " + this.context + " because of error.", e); + LockRegistryLeaderInitiator.this.future = + LockRegistryLeaderInitiator.this.executorService.submit( + () -> { + // Give it a chance to elect some other leader. + Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis); + return call(); + }); + } + return true; + } + else { + if (isRunning()) { + // Give it a chance to elect some other leader. + try { + Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis); + } + catch (InterruptedException e1) { + // Ignore interruption and let it to be caught on the next cycle. + Thread.currentThread().interrupt(); + } + } + if (logger.isDebugEnabled()) { + logger.debug("Error acquiring the lock for " + this.context + + ". " + (isRunning() ? "Retrying..." : ""), e); + } + } + return false; + } + public boolean isLeader() { return this.locked; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/SyslogToMapTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/SyslogToMapTransformer.java index 280bf8663a..0153475464 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/SyslogToMapTransformer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/SyslogToMapTransformer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,7 +16,7 @@ package org.springframework.integration.transformer; -import java.io.UnsupportedEncodingException; +import java.nio.charset.StandardCharsets; import java.sql.Date; import java.time.LocalDate; import java.time.format.DateTimeFormatter; @@ -38,13 +38,12 @@ import org.springframework.util.StringUtils; * @author Gary Russell * @author Artem Bilan * @author Karol Dowbecki + * * @since 2.2 * */ public class SyslogToMapTransformer extends AbstractPayloadTransformer> { - private final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("MMM dd HH:mm:ss"); - public static final String FACILITY = "FACILITY"; public static final String SEVERITY = "SEVERITY"; @@ -59,6 +58,8 @@ public class SyslogToMapTransformer extends AbstractPayloadTransformer]+)>(.{15}) ([^ ]+) ([a-zA-Z0-9]{0,32})(.*)", Pattern.DOTALL); @Override @@ -69,85 +70,86 @@ public class SyslogToMapTransformer extends AbstractPayloadTransformer transform(byte[] payloadBytes) { - String payload; - try { - payload = new String(payloadBytes, "UTF-8"); - } - catch (@SuppressWarnings("unused") UnsupportedEncodingException e) { - payload = new String(payloadBytes); - } - return transform(payload); + return transform(new String(payloadBytes, StandardCharsets.UTF_8)); } private Map transform(String payload) { - Map map = new LinkedHashMap(); + Map map = new LinkedHashMap<>(); Matcher matcher = this.pattern.matcher(payload); if (matcher.matches()) { - try { - String facilityString = matcher.group(1); - int facility = Integer.parseInt(facilityString); - int severity = facility & 0x7; - facility = facility >> 3; - map.put(FACILITY, facility); - map.put(SEVERITY, severity); - String timestamp = matcher.group(2); - try { - LocalDate localDate = this.dateTimeFormatter.parse(timestamp, LocalDate::from); - Calendar calendar = Calendar.getInstance(); - int year = calendar.get(Calendar.YEAR); - int month = calendar.get(Calendar.MONTH); - calendar.setTime(Date.valueOf(localDate)); - /* - * syslog date doesn't include a year so we - * need to insert the current year - adjusted - * if necessary if close to midnight on Dec 31. - */ - if (month == 11 && calendar.get(Calendar.MONTH) == 0) { - calendar.set(Calendar.YEAR, year + 1); - } - else if (month == 0 && calendar.get(Calendar.MONTH) == 1) { - calendar.set(Calendar.YEAR, year - 1); - } - else { - calendar.set(Calendar.YEAR, year); - } - map.put(TIMESTAMP, calendar.getTime()); - } - catch (@SuppressWarnings("unused") Exception e) { - /* - * If we can't parse the timestamp, return it as an - * unmodified String. (Postel's law). - */ - map.put(TIMESTAMP, timestamp); - } - map.put(HOST, matcher.group(3)); - if (StringUtils.hasLength(matcher.group(4))) { - map.put(TAG, matcher.group(4)); - } - map.put(MESSAGE, matcher.group(5)); - } - catch (Exception e) { - if (logger.isDebugEnabled()) { - logger.debug("Could not decode:" + payload, e); - } - map.clear(); - map.put(UNDECODED, payload); - } + parseMatcherToMap(payload, matcher, map); } else { if (logger.isDebugEnabled()) { - logger.debug("Could not decode:" + payload); + logger.debug("Could not decode: " + payload); } map.put(UNDECODED, payload); } return map; } + private void parseMatcherToMap(Object payload, Matcher matcher, Map map) { + try { + String facilityString = matcher.group(1); + int facility = Integer.parseInt(facilityString); + int severity = facility & 0x7; + facility = facility >> 3; + map.put(FACILITY, facility); + map.put(SEVERITY, severity); + String timestamp = matcher.group(2); + parseTimestampToMap(timestamp, map); + map.put(HOST, matcher.group(3)); + String tag = matcher.group(4); + if (StringUtils.hasLength(tag)) { + map.put(TAG, tag); + } + map.put(MESSAGE, matcher.group(5)); + } + catch (Exception e) { + if (logger.isDebugEnabled()) { + logger.debug("Could not decode: " + payload, e); + } + map.clear(); + map.put(UNDECODED, payload); + } + } + + private void parseTimestampToMap(String timestamp, Map map) { + try { + LocalDate localDate = this.dateTimeFormatter.parse(timestamp, LocalDate::from); + Calendar calendar = Calendar.getInstance(); + int year = calendar.get(Calendar.YEAR); + int month = calendar.get(Calendar.MONTH); + calendar.setTime(Date.valueOf(localDate)); + /* + * syslog date doesn't include a year so we + * need to insert the current year - adjusted + * if necessary if close to midnight on Dec 31. + */ + if (month == 11 && calendar.get(Calendar.MONTH) == Calendar.JANUARY) { + calendar.set(Calendar.YEAR, year + 1); + } + else if (month == 0 && calendar.get(Calendar.MONTH) == Calendar.FEBRUARY) { + calendar.set(Calendar.YEAR, year - 1); + } + else { + calendar.set(Calendar.YEAR, year); + } + map.put(TIMESTAMP, calendar.getTime()); + } + catch (@SuppressWarnings("unused") Exception e) { + /* + * If we can't parse the timestamp, return it as an + * unmodified String. (Postel's law). + */ + map.put(TIMESTAMP, timestamp); + } + } + }