* One thing to keep in mind, however, is that any active transactional context will not - * propagate from the original sender to the eventual recipient. This is a side-effect of + * propagate from the original sender to the eventual recipient. This is a side effect of * passing the Message to the output channel after the delay with a different Thread in * control. *
@@ -532,7 +532,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement protected void rescheduleAt(Message message, Date startTime) { Runnable releaseTask = releaseTaskForMessage(message); - getTaskScheduler().schedule(releaseTask, startTime); + getTaskScheduler().schedule(releaseTask, startTime.toInstant()); } private void doReleaseMessage(Message message) { @@ -594,7 +594,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement else { releaseMessage(message); } - }, new Date())); + }, Instant.now())); } } 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 3c56630c00..87eb8ad2ca 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2021 the original author or authors. + * Copyright 2002-2022 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. @@ -31,6 +31,7 @@ import java.nio.charset.Charset; import java.nio.file.Files; import java.nio.file.StandardCopyOption; import java.nio.file.attribute.PosixFilePermission; +import java.time.Duration; import java.util.BitSet; import java.util.HashMap; import java.util.Iterator; @@ -450,7 +451,8 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand 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); // NOSONAR + this.flushTask = taskScheduler + .scheduleAtFixedRate(new Flusher(), Duration.ofMillis(this.flushInterval / 3)); // NOSONAR } } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java index 1fed68ba55..78cca97ef0 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java @@ -17,6 +17,8 @@ package org.springframework.integration.file.tail; import java.io.File; +import java.io.Serial; +import java.time.Duration; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.atomic.AtomicLong; @@ -37,6 +39,7 @@ import org.springframework.util.Assert; * @author Artem Bilan * @author Ali Shahbour * @author Vladimir Plizga + * * @since 3.0 * */ @@ -146,21 +149,21 @@ public abstract class FileTailingMessageProducerSupport extends MessageProducerS } } - - @Override protected void doStart() { if (this.idleEventInterval > 0) { - this.idleEventScheduledFuture = getTaskScheduler().scheduleWithFixedDelay(() -> { - long now = System.currentTimeMillis(); - long lastAlertAt = this.lastNoMessageAlert.get(); - long lastSend = this.lastProduce; - if (now > lastSend + this.idleEventInterval - && now > lastAlertAt + this.idleEventInterval - && this.lastNoMessageAlert.compareAndSet(lastAlertAt, now)) { - publishIdleEvent(now - lastSend); - } - }, this.idleEventInterval); + this.idleEventScheduledFuture = + getTaskScheduler() + .scheduleWithFixedDelay(() -> { + long now = System.currentTimeMillis(); + long lastAlertAt = this.lastNoMessageAlert.get(); + long lastSend = this.lastProduce; + if (now > lastSend + this.idleEventInterval + && now > lastAlertAt + this.idleEventInterval + && this.lastNoMessageAlert.compareAndSet(lastAlertAt, now)) { + publishIdleEvent(now - lastSend); + } + }, Duration.ofMillis(this.idleEventInterval)); } } @@ -191,6 +194,7 @@ public abstract class FileTailingMessageProducerSupport extends MessageProducerS public static class FileTailingIdleEvent extends FileTailingEvent { + @Serial private static final long serialVersionUID = -967118535347976767L; private final long idleTime; @@ -210,6 +214,7 @@ public abstract class FileTailingMessageProducerSupport extends MessageProducerS public static class FileTailingEvent extends FileIntegrationEvent { + @Serial private static final long serialVersionUID = -3382255736225946206L; private final String message; diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java index 4238fd295b..15ff555306 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java @@ -19,6 +19,7 @@ package org.springframework.integration.file.tail; import java.io.BufferedReader; import java.io.IOException; import java.io.InputStreamReader; +import java.time.Instant; import java.util.Date; import org.springframework.messaging.MessagingException; @@ -165,9 +166,10 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr destroyProcess(); } if (isRunning()) { - logger.info(() -> "Restarting tail process in " + getMissingFileDelay() + " milliseconds"); + long missingFileDelay = getMissingFileDelay(); + logger.info(() -> "Restarting tail process in " + missingFileDelay + " milliseconds"); getTaskScheduler() - .schedule(this::runExec, new Date(System.currentTimeMillis() + getMissingFileDelay())); + .schedule(this::runExec, Instant.now().plusMillis(missingFileDelay)); } }); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java index d9042c81ab..cccd670015 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2001-2021 the original author or authors. + * Copyright 2001-2022 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. @@ -17,7 +17,7 @@ package org.springframework.integration.ip.tcp; import java.io.IOException; -import java.util.Date; +import java.time.Instant; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; @@ -440,11 +440,12 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler this.connection = connection; this.haveSemaphore = haveSemaphore; if (async && remoteTimeout > 0) { - getTaskScheduler().schedule(() -> { - TcpOutboundGateway.this.pendingReplies.remove(connection.getConnectionId()); - this.future.setException( - new MessageTimeoutException(requestMessage, "Timed out waiting for response")); - }, new Date(System.currentTimeMillis() + remoteTimeout)); + getTaskScheduler() + .schedule(() -> { + TcpOutboundGateway.this.pendingReplies.remove(connection.getConnectionId()); + this.future.setException( + new MessageTimeoutException(requestMessage, "Timed out waiting for response")); + }, Instant.now().plusMillis(remoteTimeout)); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java index 8a865c730f..4435d5b79f 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2022 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,6 +16,7 @@ package org.springframework.integration.ip.tcp; +import java.time.Duration; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.atomic.AtomicInteger; @@ -72,7 +73,7 @@ public class TcpReceivingChannelAdapter boolean isErrorMessage = message instanceof ErrorMessage; try { if (this.shuttingDown) { - logger.info(() -> "Inbound message ignored; shutting down; " + message.toString()); + logger.info(() -> "Inbound message ignored; shutting down; " + message); } else { if (isErrorMessage) { @@ -134,7 +135,7 @@ public class TcpReceivingChannelAdapter this.clientModeConnectionManager = manager; TaskScheduler taskScheduler = getTaskScheduler(); Assert.state(taskScheduler != null, "Client mode requires a task scheduler"); - this.scheduledFuture = taskScheduler.scheduleAtFixedRate(manager, this.retryInterval); + this.scheduledFuture = taskScheduler.scheduleAtFixedRate(manager, Duration.ofMillis(this.retryInterval)); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java index 0dde105c92..15b3d19521 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2022 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. @@ -17,6 +17,7 @@ package org.springframework.integration.ip.tcp; import java.io.IOException; +import java.time.Duration; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ScheduledFuture; @@ -267,7 +268,8 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements this.clientModeConnectionManager = manager; TaskScheduler taskScheduler = getTaskScheduler(); Assert.state(taskScheduler != null, "Client mode requires a task scheduler"); - this.scheduledFuture = taskScheduler.scheduleAtFixedRate(manager, this.retryInterval); + this.scheduledFuture = + taskScheduler.scheduleAtFixedRate(manager, Duration.ofMillis(this.retryInterval)); } } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java index a3b6df1263..6080df8960 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2001-2020 the original author or authors. + * Copyright 2001-2022 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. @@ -20,7 +20,7 @@ import java.net.ServerSocket; import java.net.Socket; import java.net.SocketAddress; import java.net.SocketException; -import java.util.Date; +import java.time.Instant; import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.task.TaskRejectedException; @@ -56,7 +56,6 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection /** * The port on which the factory will listen. - * * @param port The port. */ public AbstractServerConnectionFactory(int port) { @@ -201,10 +200,10 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection return 0; } - protected void publishServerExceptionEvent(Exception e) { + protected void publishServerExceptionEvent(Exception ex) { ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher(); if (applicationEventPublisher != null) { - applicationEventPublisher.publishEvent(new TcpConnectionServerExceptionEvent(this, e)); + applicationEventPublisher.publishEvent(new TcpConnectionServerExceptionEvent(this, ex)); } } @@ -215,7 +214,7 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection TaskScheduler taskScheduler = getTaskScheduler(); if (taskScheduler != null) { try { - taskScheduler.schedule(() -> eventPublisher.publishEvent(event), new Date()); + taskScheduler.schedule(() -> eventPublisher.publishEvent(event), Instant.now()); } catch (@SuppressWarnings("unused") TaskRejectedException e) { eventPublisher.publishEvent(event);