diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultHeaderChannelRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultHeaderChannelRegistry.java index b2a1f57ee0..e60a1b864e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultHeaderChannelRegistry.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultHeaderChannelRegistry.java @@ -41,6 +41,7 @@ import org.springframework.util.Assert; * The actual average expiry time will be 1.5x the delay. * * @author Gary Russell + * @author Artem Bilan * @since 3.0 * */ @@ -135,6 +136,7 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport this.running = false; if (this.reaperScheduledFuture != null) { this.reaperScheduledFuture.cancel(true); + this.reaperScheduledFuture = null; } this.explicitlyStopped = true; } @@ -198,6 +200,7 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport public synchronized void runReaper() { if (this.reaperScheduledFuture != null) { this.reaperScheduledFuture.cancel(true); + this.reaperScheduledFuture = null; } this.run(); } 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 883d51b46e..82588a6e9c 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 @@ -54,6 +54,7 @@ import org.springframework.util.Assert; * be useful. * * @author Dave Syer + * @author Artem Bilan * @since 4.3.1 */ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBean, ApplicationEventPublisherAware { @@ -277,7 +278,10 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe synchronized (this.lifecycleMonitor) { if (this.running) { this.running = false; - this.future.cancel(true); + if (this.future != null) { + this.future.cancel(true); + } + this.future = null; logger.debug("Stopped LeaderInitiator"); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadDeserializingTransformerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadDeserializingTransformerParserTests.java index 250face2a2..40c3e4c880 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadDeserializingTransformerParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadDeserializingTransformerParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2016 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. @@ -65,7 +65,7 @@ public class PayloadDeserializingTransformerParserTests { public void directChannelWithSerializedStringMessage() throws Exception { byte[] bytes = serialize("foo"); directInput.send(new GenericMessage(bytes)); - Message result = output.receive(0); + Message result = output.receive(10000); assertNotNull(result); assertTrue(result.getPayload() instanceof String); assertEquals("foo", result.getPayload()); @@ -75,7 +75,7 @@ public class PayloadDeserializingTransformerParserTests { public void queueChannelWithSerializedStringMessage() throws Exception { byte[] bytes = serialize("foo"); queueInput.send(new GenericMessage(bytes)); - Message result = output.receive(3000); + Message result = output.receive(10000); assertNotNull(result); assertTrue(result.getPayload() instanceof String); assertEquals("foo", result.getPayload()); @@ -85,7 +85,7 @@ public class PayloadDeserializingTransformerParserTests { public void directChannelWithSerializedObjectMessage() throws Exception { byte[] bytes = serialize(new TestBean()); directInput.send(new GenericMessage(bytes)); - Message result = output.receive(0); + Message result = output.receive(10000); assertNotNull(result); assertEquals(TestBean.class, result.getPayload().getClass()); assertEquals("test", ((TestBean) result.getPayload()).name); @@ -95,7 +95,7 @@ public class PayloadDeserializingTransformerParserTests { public void queueChannelWithSerializedObjectMessage() throws Exception { byte[] bytes = serialize(new TestBean()); queueInput.send(new GenericMessage(bytes)); - Message result = output.receive(3000); + Message result = output.receive(10000); assertNotNull(result); assertEquals(TestBean.class, result.getPayload().getClass()); assertEquals("test", ((TestBean) result.getPayload()).name); @@ -110,7 +110,7 @@ public class PayloadDeserializingTransformerParserTests { @Test public void customDeserializer() throws Exception { customDeserializerInput.send(new GenericMessage("test".getBytes("UTF-8"))); - Message result = output.receive(3000); + Message result = output.receive(10000); assertNotNull(result); assertEquals(String.class, result.getPayload().getClass()); assertEquals("TEST", result.getPayload()); @@ -138,6 +138,7 @@ public class PayloadDeserializingTransformerParserTests { public Object deserialize(InputStream source) throws IOException { return FileCopyUtils.copyToString(new InputStreamReader(source, "UTF-8")).toUpperCase(); } + } } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java index 0a065d7dc1..1b22eaff57 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java @@ -38,6 +38,7 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.PriorityBlockingQueue; +import java.util.concurrent.atomic.AtomicBoolean; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -87,6 +88,8 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement private static final Log logger = LogFactory.getLog(FileReadingMessageSource.class); + private final AtomicBoolean running = new AtomicBoolean(); + /* * {@link PriorityBlockingQueue#iterator()} throws * {@link java.util.ConcurrentModificationException} in Java 5. @@ -104,8 +107,6 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement private volatile boolean scanEachPoll = false; - private volatile boolean running; - private FileListFilter filter; private FileLocker locker; @@ -281,7 +282,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement public void setWatchEvents(WatchEventType... watchEvents) { Assert.notEmpty(watchEvents, "'watchEvents' must not be empty."); Assert.noNullElements(watchEvents, "'watchEvents' must not contain null elements."); - Assert.state(!this.running, "Cannot change watch events while running."); + Assert.state(!this.running.get(), "Cannot change watch events while running."); this.watchEvents = Arrays.copyOf(watchEvents, watchEvents.length); } @@ -293,23 +294,21 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement @Override public void start() { - if (this.scanner instanceof Lifecycle) { + if (!this.running.getAndSet(true) && this.scanner instanceof Lifecycle) { ((Lifecycle) this.scanner).start(); } - this.running = true; } @Override public void stop() { - if (this.scanner instanceof Lifecycle) { - ((Lifecycle) this.scanner).start(); + if (this.running.getAndSet(false) && this.scanner instanceof Lifecycle) { + ((Lifecycle) this.scanner).stop(); } - this.running = false; } @Override public boolean isRunning() { - return this.running; + return this.running.get(); } @Override @@ -444,6 +443,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement try { this.watcher.close(); this.watcher = null; + this.pathKeys.clear(); } catch (IOException e) { logger.error("Failed to close watcher for " + FileReadingMessageSource.this.directory, e); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java index b46d1e5bef..340b226f7d 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java @@ -364,7 +364,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand @Override public void start() { - if (FileExistsMode.APPEND_NO_FLUSH.equals(this.fileExistsMode)) { + if (this.flushTask == null && FileExistsMode.APPEND_NO_FLUSH.equals(this.fileExistsMode)) { TaskScheduler taskScheduler = getTaskScheduler(); Assert.state(taskScheduler != null, "'taskScheduler' is required for FileExistsMode.APPEND_NO_FLUSH"); this.flushTask = taskScheduler.scheduleAtFixedRate(new Flusher(), this.flushInterval / 3); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java index d529090d1d..12c9926aaa 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java @@ -105,6 +105,7 @@ public class WatchServiceDirectoryScannerTests { assertTrue(files.contains(top1)); assertTrue(files.contains(foo1)); assertTrue(files.contains(bar1)); + fileReadingMessageSource.start(); File top2 = this.folder.newFile(); File foo2 = File.createTempFile("foo", ".txt", this.foo); File bar2 = File.createTempFile("bar", ".txt", this.bar); @@ -130,6 +131,7 @@ public class WatchServiceDirectoryScannerTests { var1 = StandardWatchEventKinds.OVERFLOW; } */ + fileReadingMessageSource.start(); List filesForOverflow = new ArrayList(600); for (int i = 0; i < 600; i++) { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java index 15f9bfe291..caf26eee8d 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java @@ -197,7 +197,7 @@ public class UdpChannelAdapterTests { e.printStackTrace(); } }); - Message receivedMessage = (Message) channel.receive(2000); + Message receivedMessage = (Message) channel.receive(10000); assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload())); String replyString = "reply:" + System.currentTimeMillis(); byte[] replyBytes = replyString.getBytes(); @@ -241,7 +241,7 @@ public class UdpChannelAdapterTests { handler.start(); Message message = MessageBuilder.withPayload("ABCD".getBytes()).build(); handler.handleMessage(message); - Message receivedMessage = (Message) channel.receive(2000); + Message receivedMessage = (Message) channel.receive(10000); assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload())); adapter.stop(); handler.stop(); @@ -268,7 +268,7 @@ public class UdpChannelAdapterTests { datagramSocket.send(packet); datagramSocket.close(); - Message receivedMessage = (Message) channel.receive(2000); + Message receivedMessage = (Message) channel.receive(10000); assertNotNull(receivedMessage); assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload())); adapter.stop(); @@ -292,7 +292,7 @@ public class UdpChannelAdapterTests { Message message = MessageBuilder.withPayload("ABCD".getBytes()).build(); handler.handleMessage(message); - Message receivedMessage = (Message) channel.receive(2000); + Message receivedMessage = (Message) channel.receive(10000); assertNotNull(receivedMessage); assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload())); adapter.stop(); @@ -320,7 +320,7 @@ public class UdpChannelAdapterTests { DatagramSocket datagramSocket = new DatagramSocket(0); datagramSocket.send(packet); datagramSocket.close(); - Message receivedMessage = errorChannel.receive(2000); + Message receivedMessage = errorChannel.receive(10000); assertNotNull(receivedMessage); assertEquals("Failed", ((Exception) receivedMessage.getPayload()).getCause().getMessage()); adapter.stop(); diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java index e88c3e99ae..c2152fd682 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java @@ -16,6 +16,8 @@ package org.springframework.integration.mqtt.outbound; +import java.util.concurrent.atomic.AtomicBoolean; + import org.springframework.context.Lifecycle; import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; @@ -29,11 +31,14 @@ import org.springframework.util.Assert; * Abstract class for MQTT outbound channel adapters. * * @author Gary Russell + * @author Artem Bilan * @since 4.0 * */ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler implements Lifecycle { + private final AtomicBoolean running = new AtomicBoolean(); + private final String url; private final String clientId; @@ -46,8 +51,6 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler private volatile MessageConverter converter; - private boolean running; - private volatile int clientInstance; public AbstractMqttMessageHandler(String url, String clientId) { @@ -113,23 +116,25 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler @Override public final void start() { - this.doStart(); - this.running = true; + if (!this.running.getAndSet(true)) { + doStart(); + } } protected abstract void doStart(); @Override public final void stop() { - this.doStop(); - this.running = false; + if (this.running.getAndSet(false)) { + doStop(); + } } protected abstract void doStop(); @Override public boolean isRunning() { - return this.running; + return this.running.get(); } @Override diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java index 248bda02e9..aa0b53fb60 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java @@ -38,6 +38,7 @@ import org.springframework.util.Assert; * Eclipse Paho implementation. * * @author Gary Russell + * @author Artem Bilan * @since 4.0 * */ @@ -144,9 +145,10 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler @Override protected void doStop() { try { - if (this.client != null) { - this.client.disconnect().waitForCompletion(this.completionTimeout); - this.client.close(); + IMqttAsyncClient client = this.client; + if (client != null) { + client.disconnect().waitForCompletion(this.completionTimeout); + client.close(); this.client = null; } }