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

Fix more deprecations around TaskScheduler

Browse Source
This commit is contained in:
Artem Bilan
2022-07-11 11:41:38 -04:00
parent 4a55fa3188
commit f85423a436
9 changed files with 50 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultHeaderChannelRegistry.java
View File

@@ -17,7 +17,6 @@
package org.springframework.integration.channel;
import java.time.Instant;
import java.util.Date;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
@@ -224,7 +223,7 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
}
this.reaperScheduledFuture =
getTaskScheduler()
.schedule(this, new Date(System.currentTimeMillis() + this.reaperDelay));
.schedule(this, Instant.now().plusMillis(this.reaperDelay));
logger.trace(() -> "Reaper completed; channels size=" + this.channels.size());
}

6
spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java
View File

@@ -68,7 +68,7 @@ import org.springframework.util.ObjectUtils;
* concurrently, even very long delays, without producing a buildup of blocked Threads.
* <p>
* 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.
* <p>
@@ -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()));
}
}

6
spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java
View File

@@ -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
}
}

29
spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java
View File

@@ -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;

6
spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java
View File

@@ -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));
}
});
}

15
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java
View File

@@ -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));
}
}

7
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java
View File

@@ -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));
}
}

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java
View File

@@ -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));
}
}
}

11
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java
View File

@@ -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);