From e259162527409940dae3576f22454682a711e9b4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=9D=D0=B8=D0=BA=D0=BE=D0=BB=D0=B0=D0=B9=20=D0=91=D0=B5?= =?UTF-8?q?=D0=BB=D0=BE=D0=BD=D0=BE=D0=B3=D0=BE=D0=B2?= Date: Sat, 17 Oct 2026 00:57:27 +0300 Subject: [PATCH] refactor(core): refactor timeouts in core module Change timeout type from millis long to Duration --- .../factory/IProtoCallClusterSpace.java | 3 +- .../client/factory/TarantoolBoxSpaceImpl.java | 13 +- .../client/factory/TarantoolClientImpl.java | 3 +- .../benchmark/CustomBenchmarkRunner.java | 11 +- .../java/io/tarantool/core/IProtoClient.java | 39 +- .../io/tarantool/core/IProtoClientImpl.java | 60 +- .../java/io/tarantool/core/TimeoutsUtils.java | 60 ++ .../io/tarantool/core/WatcherOptions.java | 62 ++- .../tarantool/core/connection/Connection.java | 15 + .../ConnectionChannelInitializer.java | 19 +- .../core/connection/ConnectionImpl.java | 54 +- .../core/protocol/IProtoRequestOpts.java | 31 +- .../protocol/fsm/RequestStateMachine.java | 17 +- .../protocol/fsm/WatcherStateMachine.java | 4 +- .../core/protocol/requests/IProtoBegin.java | 25 +- .../tarantool/core/integration/BaseTest.java | 3 +- .../ConnectionCloseOnClientSideTest.java | 5 +- .../ConnectionCloseOnServerSideTest.java | 3 +- .../ConnectionToTarantoolTest.java | 516 ++++++++++-------- .../integration/GracefulShutdownTest.java | 3 +- .../core/integration/IProtoClientTest.java | 17 +- .../integration/IProtoClientWatchersTest.java | 5 +- .../core/integration/MVCCStreamsTest.java | 29 +- .../protocol/fsm/WatcherStateMachineTest.java | 13 +- .../core/unit/requests/RequestTest.java | 3 +- .../benchmark/CustomBenchmarkRunner.java | 8 +- .../mapping/integration/IProtoClientTest.java | 3 +- .../java/io/tarantool/pool/PoolEntry.java | 17 +- .../pool/integration/ConnectionPoolTest.java | 3 +- .../TarantoolSchemaFetcherTest.java | 7 +- 30 files changed, 677 insertions(+), 374 deletions(-) create mode 100644 tarantool-core/src/main/java/io/tarantool/core/TimeoutsUtils.java diff --git a/tarantool-client/src/main/java/io/tarantool/client/factory/IProtoCallClusterSpace.java b/tarantool-client/src/main/java/io/tarantool/client/factory/IProtoCallClusterSpace.java index 02d4e6f6..5ea0e036 100644 --- a/tarantool-client/src/main/java/io/tarantool/client/factory/IProtoCallClusterSpace.java +++ b/tarantool-client/src/main/java/io/tarantool/client/factory/IProtoCallClusterSpace.java @@ -5,6 +5,7 @@ package io.tarantool.client.factory; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -78,7 +79,7 @@ public CompletableFuture iprotoCall( TarantoolJacksonMapping.toValue(withSpaceName(args)), null, IProtoRequestOpts.empty() - .withRequestTimeout(options.getTimeout()) + .withRequestTimeout(Duration.ofMillis(options.getTimeout())) .withStreamId(options.getStreamId()))); } diff --git a/tarantool-client/src/main/java/io/tarantool/client/factory/TarantoolBoxSpaceImpl.java b/tarantool-client/src/main/java/io/tarantool/client/factory/TarantoolBoxSpaceImpl.java index 7e3ebdbc..5a5e986d 100644 --- a/tarantool-client/src/main/java/io/tarantool/client/factory/TarantoolBoxSpaceImpl.java +++ b/tarantool-client/src/main/java/io/tarantool/client/factory/TarantoolBoxSpaceImpl.java @@ -5,6 +5,7 @@ package io.tarantool.client.factory; +import java.time.Duration; import java.util.Arrays; import java.util.List; import java.util.Objects; @@ -441,7 +442,7 @@ private CompletableFuture iprotoUpsert( client -> { final IProtoRequestOpts requestOpts = IProtoRequestOpts.empty() - .withRequestTimeout(options.getTimeout()) + .withRequestTimeout(Duration.ofMillis(options.getTimeout())) .withStreamId(options.getStreamId()); int indexId; @@ -496,7 +497,7 @@ private CompletableFuture iprotoUpdate( client -> { final IProtoRequestOpts requestOpts = IProtoRequestOpts.empty() - .withRequestTimeout(options.getTimeout()) + .withRequestTimeout(Duration.ofMillis(options.getTimeout())) .withStreamId(options.getStreamId()); Integer indexId; @@ -548,7 +549,7 @@ private CompletableFuture iprotoDelete(List key, DeleteOption client -> { final IProtoRequestOpts requestOpts = IProtoRequestOpts.empty() - .withRequestTimeout(options.getTimeout()) + .withRequestTimeout(Duration.ofMillis(options.getTimeout())) .withStreamId(options.getStreamId()); Integer indexId; @@ -613,7 +614,7 @@ private CompletableFuture iprotoSelect(List key, SelectOption client -> { final IProtoRequestOpts requestOpts = IProtoRequestOpts.empty() - .withRequestTimeout(options.getTimeout()) + .withRequestTimeout(Duration.ofMillis(options.getTimeout())) .withStreamId(options.getStreamId()); Integer indexId; @@ -673,7 +674,7 @@ private CompletableFuture iprotoReplace(Object tuple, Options op client -> { final IProtoRequestOpts requestOpts = IProtoRequestOpts.empty() - .withRequestTimeout(options.getTimeout()) + .withRequestTimeout(Duration.ofMillis(options.getTimeout())) .withStreamId(options.getStreamId()); if (fetcher == null) { @@ -715,7 +716,7 @@ private CompletableFuture iprotoInsert(Object tuple, Options opt client -> { final IProtoRequestOpts requestOpts = IProtoRequestOpts.empty() - .withRequestTimeout(options.getTimeout()) + .withRequestTimeout(Duration.ofMillis(options.getTimeout())) .withStreamId(options.getStreamId()); if (fetcher == null) { diff --git a/tarantool-client/src/main/java/io/tarantool/client/factory/TarantoolClientImpl.java b/tarantool-client/src/main/java/io/tarantool/client/factory/TarantoolClientImpl.java index d4a1a679..fca01fbf 100644 --- a/tarantool-client/src/main/java/io/tarantool/client/factory/TarantoolClientImpl.java +++ b/tarantool-client/src/main/java/io/tarantool/client/factory/TarantoolClientImpl.java @@ -7,6 +7,7 @@ import java.lang.reflect.Constructor; import java.lang.reflect.InvocationTargetException; +import java.time.Duration; import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; @@ -168,7 +169,7 @@ protected TarantoolClientImpl( */ private IProtoRequestOpts convertOptions(Options opts) { return IProtoRequestOpts.empty() - .withRequestTimeout(opts.getTimeout()) + .withRequestTimeout(Duration.ofMillis(opts.getTimeout())) .withStreamId(opts.getStreamId()); } diff --git a/tarantool-client/src/test/java/io/tarantool/client/benchmark/CustomBenchmarkRunner.java b/tarantool-client/src/test/java/io/tarantool/client/benchmark/CustomBenchmarkRunner.java index e7c80f03..18cf0e18 100644 --- a/tarantool-client/src/test/java/io/tarantool/client/benchmark/CustomBenchmarkRunner.java +++ b/tarantool-client/src/test/java/io/tarantool/client/benchmark/CustomBenchmarkRunner.java @@ -6,6 +6,7 @@ package io.tarantool.client.benchmark; import java.math.BigDecimal; +import java.time.Duration; import java.time.Instant; import java.time.LocalDateTime; import java.time.ZoneOffset; @@ -189,7 +190,7 @@ public static Boolean beginReplaceCommit( () -> { long streamId = iprotoClient.allocateStreamId(); return iprotoClient - .begin(streamId, 2_000, TransactionIsolationLevel.DEFAULT) + .begin(streamId, Duration.ofSeconds(2), TransactionIsolationLevel.DEFAULT) .thenCompose( r -> client @@ -197,7 +198,8 @@ public static Boolean beginReplaceCommit( .replace( Arrays.asList(counter.getAndIncrement(), "artyom"), BaseOptions.builder().withStreamId(streamId).build())) - .thenCompose(r -> iprotoClient.commit(streamId)); + .thenCompose(r -> iprotoClient.commit(streamId)) + .toCompletableFuture(); }); } @@ -221,7 +223,7 @@ public static Boolean beginThreeReplaceCommit( () -> { long streamId = iprotoClient.allocateStreamId(); return iprotoClient - .begin(streamId, 2_000, TransactionIsolationLevel.DEFAULT) + .begin(streamId, Duration.ofSeconds(2), TransactionIsolationLevel.DEFAULT) .thenCompose( r -> client @@ -243,7 +245,8 @@ public static Boolean beginThreeReplaceCommit( .replace( Arrays.asList(counter.getAndIncrement(), "artyom"), BaseOptions.builder().withStreamId(streamId).build())) - .thenCompose(r -> iprotoClient.commit(streamId)); + .thenCompose(r -> iprotoClient.commit(streamId)) + .toCompletableFuture(); }); } diff --git a/tarantool-core/src/main/java/io/tarantool/core/IProtoClient.java b/tarantool-core/src/main/java/io/tarantool/core/IProtoClient.java index e60c235a..e850971d 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/IProtoClient.java +++ b/tarantool-core/src/main/java/io/tarantool/core/IProtoClient.java @@ -6,9 +6,11 @@ package io.tarantool.core; import java.net.InetSocketAddress; +import java.time.Duration; import java.util.List; import java.util.Set; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; import java.util.function.BiConsumer; import java.util.function.Consumer; @@ -25,11 +27,24 @@ public interface IProtoClient { + /** + * @deprecated Use {@link #connect(InetSocketAddress, Duration)} + */ + @Deprecated CompletableFuture connect(InetSocketAddress address, long timeoutMs); + CompletionStage connect(InetSocketAddress address, Duration timeout); + + /** + * @deprecated Use {@link #connect(InetSocketAddress, Duration, boolean)} + */ + @Deprecated CompletableFuture connect( InetSocketAddress address, long timeoutMs, boolean gracefulShutdown); + CompletionStage connect( + InetSocketAddress address, Duration timeout, boolean gracefulShutdown); + CompletableFuture authorize(String user, String password); CompletableFuture authorize(String user, String password, IProtoRequestOpts opts); @@ -201,11 +216,25 @@ CompletableFuture execute( long allocateStreamId(); + /** + * @deprecated Use {@link #begin(Long, Duration, TransactionIsolationLevel)} + */ + @Deprecated CompletableFuture begin( - Long streamId, long timeout, TransactionIsolationLevel level); + Long streamId, long timeoutMs, TransactionIsolationLevel level); + + CompletionStage begin( + Long streamId, Duration timeout, TransactionIsolationLevel level); + /** + * @deprecated Use {@link #begin(Long, Duration, TransactionIsolationLevel, IProtoRequestOpts)} + */ + @Deprecated CompletableFuture begin( - Long streamId, long timeout, TransactionIsolationLevel level, IProtoRequestOpts opts); + Long streamId, long timeoutMs, TransactionIsolationLevel level, IProtoRequestOpts opts); + + CompletionStage begin( + Long streamId, Duration timeout, TransactionIsolationLevel level, IProtoRequestOpts opts); CompletableFuture commit(Long streamId); @@ -264,7 +293,13 @@ CompletableFuture id( void resume(); + /** + * @deprecated Use {@link #setIdleTimeout(Duration)} + */ + @Deprecated void setIdleTimeout(int idleTimeout); + void setIdleTimeout(Duration timeout); + boolean isPaused(); } diff --git a/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java b/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java index b248c732..14f08ee4 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java +++ b/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java @@ -6,6 +6,7 @@ package io.tarantool.core; import java.net.InetSocketAddress; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -15,6 +16,7 @@ import java.util.Optional; import java.util.Set; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicLong; import java.util.function.BiConsumer; @@ -78,7 +80,7 @@ public class IProtoClientImpl implements IProtoClient { public static final IProtoAuth.AuthType DEFAULT_AUTH_TYPE = IProtoAuth.AuthType.CHAP_SHA1; public static final IProtoRequestOpts DEFAULT_REQUEST_OPTS = - IProtoRequestOpts.empty().withRequestTimeout(5000); + IProtoRequestOpts.empty().withRequestTimeout(Duration.ofSeconds(5)); public static final WatcherOptions DEFAULT_WATCHER_OPTS = WatcherOptions.builder().build(); private static final Set FEATURES_SET_ENUM = EnumSet.allOf(IProtoFeature.class); private static final List FEATURES_LIST_INT = @@ -89,7 +91,7 @@ public class IProtoClientImpl implements IProtoClient { private static final String CALL_CONNECT_BEFORE_GETTING_SERVER_DETAILS = "Call connect before getting server details"; private static final String SHUTDOWN_EVENT_KEY = "box.shutdown"; - private static final int DEFAULT_IDLE_TIMEOUT = 1_000; + private static final Duration DEFAULT_IDLE_TIMEOUT = Duration.ofSeconds(1); private static final Logger log = LoggerFactory.getLogger(IProtoClientImpl.class); private final AtomicLong streamIdSequence; private final AtomicLong syncIdSequence; @@ -178,13 +180,26 @@ public IProtoClientImpl( } @Override + @Deprecated public CompletableFuture connect(InetSocketAddress address, long timeoutMs) { - return connect(address, timeoutMs, true); + return connect(address, Duration.ofMillis(timeoutMs), true).toCompletableFuture(); } @Override + public CompletionStage connect(InetSocketAddress address, Duration timeout) { + return connect(address, timeout, true); + } + + @Override + @Deprecated public CompletableFuture connect( InetSocketAddress address, long timeoutMs, boolean gracefulShutdown) { + return connect(address, Duration.ofMillis(timeoutMs), gracefulShutdown).toCompletableFuture(); + } + + @Override + public CompletionStage connect( + InetSocketAddress address, Duration timeout, boolean gracefulShutdown) { if (gracefulShutdown) { watch(SHUTDOWN_EVENT_KEY, this::shutdownEventCallback); } @@ -193,7 +208,7 @@ public CompletableFuture connect( serverProtocolVersion = new CompletableFuture<>(); serverFeatures = new CompletableFuture<>(); return connection - .connect(address, timeoutMs) + .connect(address, timeout) .thenRun( () -> { updateWatchers(); @@ -542,15 +557,36 @@ public long allocateStreamId() { return streamIdSequence.updateAndGet(n -> n == Long.MAX_VALUE ? 1 : n + 1); } + @Deprecated @Override public CompletableFuture begin( - Long streamId, long streamTimeout, TransactionIsolationLevel level) { - return begin(streamId, streamTimeout, level, DEFAULT_REQUEST_OPTS); + Long streamId, long streamTimeoutMs, TransactionIsolationLevel level) { + return begin(streamId, Duration.ofMillis(streamTimeoutMs), level, DEFAULT_REQUEST_OPTS) + .toCompletableFuture(); + } + + @Override + public CompletionStage begin( + Long streamId, Duration timeout, TransactionIsolationLevel level) { + return begin(streamId, timeout, level, DEFAULT_REQUEST_OPTS); } + @Deprecated @Override public CompletableFuture begin( - Long streamId, long streamTimeout, TransactionIsolationLevel level, IProtoRequestOpts opts) { + Long streamId, + long streamTimeoutMs, + TransactionIsolationLevel level, + IProtoRequestOpts opts) { + return begin(streamId, Duration.ofMillis(streamTimeoutMs), level, opts).toCompletableFuture(); + } + + @Override + public CompletionStage begin( + Long streamId, + Duration streamTimeout, + TransactionIsolationLevel level, + IProtoRequestOpts opts) { return runRequest(new IProtoBegin(streamId, streamTimeout, level), opts); } @@ -794,11 +830,17 @@ public void resume() { @Override public void setIdleTimeout(int idleTimeout) { - if (idleTimeout <= 0) { + setIdleTimeout(Duration.ofMillis(idleTimeout)); + } + + @Override + public void setIdleTimeout(Duration timeout) { + if (timeout.isNegative() || timeout.isZero()) { log.warn("idleTimeout must be greater than 0"); this.connection.setIdleTimeout(DEFAULT_IDLE_TIMEOUT); + return; } - this.connection.setIdleTimeout(idleTimeout); + this.connection.setIdleTimeout(timeout); } @Override diff --git a/tarantool-core/src/main/java/io/tarantool/core/TimeoutsUtils.java b/tarantool-core/src/main/java/io/tarantool/core/TimeoutsUtils.java new file mode 100644 index 00000000..827c86f8 --- /dev/null +++ b/tarantool-core/src/main/java/io/tarantool/core/TimeoutsUtils.java @@ -0,0 +1,60 @@ +/* + * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY + * All Rights Reserved. + */ + +package io.tarantool.core; + +import java.time.Duration; + +public final class TimeoutsUtils { + + private TimeoutsUtils() {} + + /** + * @see #checkTimeoutIsNotNegative(Duration, RuntimeException) + */ + public static void checkTimeoutIsNotNegative(Duration timeout) { + checkTimeoutIsNotNegative(timeout, null); + } + + /** + * Checks that the passed timeout duration is not negative ({@code timeout >= 0}) + * + * @param timeout must be not null + * @param exception custom exception + */ + public static void checkTimeoutIsNotNegative(Duration timeout, RuntimeException exception) { + if (timeout.isNegative()) { + throwException(timeout, exception); + } + } + + /** + * @see #checkTimeoutIsPositive(Duration, RuntimeException) + */ + public static void checkTimeoutIsPositive(Duration timeout) { + checkTimeoutIsPositive(timeout, null); + } + + /** + * Checks that the passed timeout duration is positive ({@code timeout > 0}). + * + * @param timeout must be not null + * @param exception custom exception + */ + public static void checkTimeoutIsPositive(Duration timeout, RuntimeException exception) { + if (timeout.isZero() || timeout.isNegative()) { + throwException(timeout, exception); + } + } + + private static void throwException(Duration timeout, RuntimeException exception) { + if (exception == null) { + throw new IllegalArgumentException( + "Timeout must be greater than 0!. Current value is '" + timeout + "'"); + } else { + throw exception; + } + } +} diff --git a/tarantool-core/src/main/java/io/tarantool/core/WatcherOptions.java b/tarantool-core/src/main/java/io/tarantool/core/WatcherOptions.java index 77b5365b..1aa9114f 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/WatcherOptions.java +++ b/tarantool-core/src/main/java/io/tarantool/core/WatcherOptions.java @@ -5,13 +5,44 @@ package io.tarantool.core; +import java.time.Duration; import java.util.function.BiConsumer; public final class WatcherOptions { + private Duration sendTimeout; + + private BiConsumer errorHandler; + + private WatcherOptions(Duration timeout, BiConsumer errback) { + this.sendTimeout = timeout; + this.errorHandler = errback; + } + + /** + * @deprecated Use {@link #sendTimeout()} instead this + */ + @Deprecated + public long getSendTimeout() { + return this.sendTimeout.toMillis(); + } + + public Duration sendTimeout() { + return this.sendTimeout; + } + + public BiConsumer getErrorHandler() { + return errorHandler; + } + + public static Builder builder() { + return new Builder(); + } + public static class Builder { - private long sendTimeout = 5000; + private Duration sendTimeout = Duration.ofSeconds(5); + private BiConsumer errorHandler = (key, err) -> {}; public Builder withErrorHandler(BiConsumer errback) { @@ -19,7 +50,16 @@ public Builder withErrorHandler(BiConsumer errback) { return this; } + /** + * @deprecated Use {@link #withSendTimeout(Duration)} instead this. + */ + @Deprecated public Builder withSendTimeout(long timeout) { + return withSendTimeout(Duration.ofMillis(timeout)); + } + + public Builder withSendTimeout(Duration timeout) { + TimeoutsUtils.checkTimeoutIsPositive(timeout); this.sendTimeout = timeout; return this; } @@ -28,24 +68,4 @@ public WatcherOptions build() { return new WatcherOptions(sendTimeout, errorHandler); } } - - private long sendTimeout; - private BiConsumer errorHandler; - - public static Builder builder() { - return new Builder(); - } - - private WatcherOptions(long timeout, BiConsumer errback) { - this.sendTimeout = timeout; - this.errorHandler = errback; - } - - public long getSendTimeout() { - return sendTimeout; - } - - public BiConsumer getErrorHandler() { - return errorHandler; - } } diff --git a/tarantool-core/src/main/java/io/tarantool/core/connection/Connection.java b/tarantool-core/src/main/java/io/tarantool/core/connection/Connection.java index 23492369..f761f648 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/connection/Connection.java +++ b/tarantool-core/src/main/java/io/tarantool/core/connection/Connection.java @@ -6,8 +6,10 @@ package io.tarantool.core.connection; import java.net.InetSocketAddress; +import java.time.Duration; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; import java.util.function.BiConsumer; import java.util.function.Consumer; @@ -16,9 +18,16 @@ public interface Connection extends AutoCloseable { + /** + * @deprecated Use {@link #connect(InetSocketAddress, Duration)} + */ + @Deprecated CompletableFuture connect(InetSocketAddress addr, long timeoutMs) throws IllegalStateException; + CompletionStage connect(InetSocketAddress address, Duration timeout) + throws IllegalStateException; + CompletableFuture send(IProtoMessage message) throws IllegalStateException; Connection listen(Consumer listener); @@ -35,7 +44,13 @@ CompletableFuture connect(InetSocketAddress addr, long timeoutMs) void resume(); + /** + * @deprecated Use {@link #setIdleTimeout(Duration)} + */ + @Deprecated void setIdleTimeout(int idleTimeout); + void setIdleTimeout(Duration timeout); + boolean isPaused(); } diff --git a/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionChannelInitializer.java b/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionChannelInitializer.java index 2d8f9c75..7497d404 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionChannelInitializer.java +++ b/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionChannelInitializer.java @@ -6,6 +6,7 @@ package io.tarantool.core.connection; import java.net.InetSocketAddress; +import java.time.Duration; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.function.BiConsumer; @@ -18,6 +19,7 @@ import io.netty.handler.ssl.SslContext; import io.netty.handler.timeout.IdleStateHandler; +import io.tarantool.core.TimeoutsUtils; import io.tarantool.core.connection.codecs.IProtoFrameDecoder; import io.tarantool.core.connection.codecs.IProtoFrameEncoder; import io.tarantool.core.connection.handlers.GreetingHandler; @@ -33,7 +35,7 @@ class ConnectionChannelInitializer extends ChannelInitializer { private final InetSocketAddress peerAddress; private final ChannelFutureListener closeHandler; private final FlushConsolidationHandler flushConsolidationHandler; - private final int idleTimeout; + private final Duration idleTimeout; private ConnectionChannelInitializer( CompletableFuture promise, @@ -42,7 +44,7 @@ private ConnectionChannelInitializer( InetSocketAddress peerAddress, ChannelFutureListener closeHandler, FlushConsolidationHandler flushConsolidationHandler, - int idleTimeout) { + Duration idleTimeout) { this.promise = promise; this.messageHandler = messageHandler; this.sslContext = sslContext; @@ -67,10 +69,10 @@ protected void initChannel(SocketChannel socketChannel) { socketChannel.alloc(), peerAddress.getHostString(), peerAddress.getPort())); } - if (this.idleTimeout > 0) { + if (this.idleTimeout.compareTo(Duration.ZERO) > 0) { pipeline.addLast( "IdleTimeoutHandler", - new IdleStateHandler(false, idleTimeout, 0, 0, TimeUnit.MILLISECONDS)); + new IdleStateHandler(false, this.idleTimeout.toNanos(), 0, 0, TimeUnit.NANOSECONDS)); } pipeline.addLast("GreetingHandler", new GreetingHandler(promise)); @@ -89,9 +91,7 @@ public static class Builder { private InetSocketAddress peerAddress = null; private ChannelFutureListener closeHandler; private FlushConsolidationHandler flushConsolidationHandler = null; - private int idleTimeout = -1; - - public Builder() {} + private Duration idleTimeout = Duration.ZERO; public Builder withConnectPromise(CompletableFuture promise) { this.promise = promise; @@ -120,8 +120,9 @@ public Builder withFlushConsolidationHandler( return this; } - public Builder withIdleTimeout(int idleTimeout) { - this.idleTimeout = idleTimeout; + public Builder withIdleTimeout(Duration timeout) { + TimeoutsUtils.checkTimeoutIsNotNegative(timeout); + this.idleTimeout = timeout; return this; } diff --git a/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionImpl.java b/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionImpl.java index 892806c4..e8b40b28 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionImpl.java +++ b/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionImpl.java @@ -6,11 +6,13 @@ package io.tarantool.core.connection; import java.net.InetSocketAddress; +import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -29,6 +31,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import io.tarantool.core.TimeoutsUtils; import io.tarantool.core.connection.exceptions.ConnectionClosedException; import io.tarantool.core.connection.exceptions.ConnectionException; import io.tarantool.core.connection.exceptions.IdleTimeoutException; @@ -118,7 +121,7 @@ public String getMessage() { private Consumer consumer; private CompletableFuture connectPromise; private Timeout timeoutHandler; - private int idleTimeout; + private Duration idleTimeout = Duration.ZERO; /** * Instantiates a new Connection. @@ -179,22 +182,29 @@ public Connection listen(Consumer listener) { } @Override + @Deprecated public synchronized CompletableFuture connect(InetSocketAddress address, long timeoutMs) throws IllegalStateException { + return connect(address, Duration.ofMillis(timeoutMs)).toCompletableFuture(); + } + + @Override + public synchronized CompletionStage connect(InetSocketAddress address, Duration timeout) + throws IllegalStateException { if (state.compareAndSet(State.CLOSED, State.CONNECTING)) { closedBy.set(CloseReason.NO_REASON); log.debug("{} is connecting to {}", this, address); CompletableFuture promise = new CompletableFuture<>(); - scheduleGreetingTimeout(promise, timeoutMs); + scheduleGreetingTimeout(promise, timeout); connectPromise = promise.thenApply( - greeting -> { + response -> { cancelGreetingTimeout(); log.info("{} connected to {}", this, address); - this.greeting = greeting; + this.greeting = response; state.set(State.READY); - return greeting; + return response; }); channel = @@ -280,11 +290,7 @@ public boolean isConnected() { @Override public Connection onClose( ConnectionCloseEvent event, BiConsumer callback) { - if (!closeListeners.containsKey(event)) { - closeListeners.put(event, new ArrayList<>()); - } - closeListeners.get(event).add(callback); - + closeListeners.computeIfAbsent(event, k -> new ArrayList<>()).add(callback); return this; } @@ -300,9 +306,19 @@ public void resume() { channel.config().setAutoRead(true); } + /** + * @deprecated Use {@link #setIdleTimeout(Duration)} instead this. + */ @Override + @Deprecated public void setIdleTimeout(int idleTimeout) { - this.idleTimeout = idleTimeout; + setIdleTimeout(Duration.ofMillis(idleTimeout)); + } + + @Override + public void setIdleTimeout(Duration timeout) { + TimeoutsUtils.checkTimeoutIsNotNegative(timeout); + this.idleTimeout = timeout; } @Override @@ -405,14 +421,14 @@ private void fire(ConnectionCloseEvent event, Throwable ex) { } } - private void scheduleGreetingTimeout(CompletableFuture promise, long timeoutMs) { - log.debug("Schedule timeout for connect in milliseconds: {}", timeoutMs); - if (timeoutMs <= 0) { - throw new IllegalStateException("Timeout must be greater than zero!"); - } + private void scheduleGreetingTimeout(CompletableFuture promise, Duration timeout) { + TimeoutsUtils.checkTimeoutIsPositive( + timeout, new IllegalStateException("Timeout must be greater than zero!")); + + log.debug("Schedule timeout for connect: {}", timeout); timeoutHandler = this.timerService.newTimeout( - timeoutHandler -> { + handler -> { if (promise.isDone()) { return; } @@ -421,8 +437,8 @@ private void scheduleGreetingTimeout(CompletableFuture promise, long t promise.completeExceptionally(new TimeoutException("Connection timeout")); } }, - timeoutMs, - TimeUnit.MILLISECONDS); + timeout.toNanos(), + TimeUnit.NANOSECONDS); } private void cancelGreetingTimeout() { diff --git a/tarantool-core/src/main/java/io/tarantool/core/protocol/IProtoRequestOpts.java b/tarantool-core/src/main/java/io/tarantool/core/protocol/IProtoRequestOpts.java index afde22ef..cc03057c 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/protocol/IProtoRequestOpts.java +++ b/tarantool-core/src/main/java/io/tarantool/core/protocol/IProtoRequestOpts.java @@ -5,28 +5,37 @@ package io.tarantool.core.protocol; +import java.time.Duration; import java.util.function.Consumer; +import io.tarantool.core.TimeoutsUtils; + public class IProtoRequestOpts { private Consumer pushHandler; private Handlers handlers; - private long requestTimeoutMs; + private Duration requestTimeout; private Long streamId; private IProtoRequestOpts() { - requestTimeoutMs = 3_000L; + this.requestTimeout = Duration.ofSeconds(3); } public static IProtoRequestOpts empty() { return new IProtoRequestOpts(); } + /** + * @deprecated Use {@link #withRequestTimeout(Duration)} + */ + @Deprecated public IProtoRequestOpts withRequestTimeout(long timeoutMs) { - if (timeoutMs <= 0) { - throw new IllegalArgumentException("timeout should be greater than 0"); - } - this.requestTimeoutMs = timeoutMs; + return withRequestTimeout(Duration.ofMillis(timeoutMs)); + } + + public IProtoRequestOpts withRequestTimeout(Duration timeout) { + TimeoutsUtils.checkTimeoutIsPositive(timeout); + this.requestTimeout = timeout; return this; } @@ -43,8 +52,16 @@ public IProtoRequestOpts withStreamId(Long streamId) { return this; } + /** + * @deprecated Use {@link #requestTimeout()} + */ + @Deprecated public long getRequestTimeout() { - return this.requestTimeoutMs; + return this.requestTimeout.toMillis(); + } + + public Duration requestTimeout() { + return this.requestTimeout; } public Long getStreamId() { diff --git a/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/RequestStateMachine.java b/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/RequestStateMachine.java index 0ce8f1f4..c6107db4 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/RequestStateMachine.java +++ b/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/RequestStateMachine.java @@ -5,6 +5,7 @@ package io.tarantool.core.protocol.fsm; +import java.time.Duration; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; @@ -30,7 +31,7 @@ public class RequestStateMachine extends AbstractIProtoStateMachine { private final Consumer pushConsumer; private final Handlers handlers; private final Timer timerService; - private final long timeout; + private final Duration timeout; private final CompletableFuture promise; private Timeout timer; @@ -48,7 +49,7 @@ public RequestStateMachine( this.pushConsumer = opts.getPushHandler(); this.handlers = requestHandlers; this.timerService = timerService; - this.timeout = opts.getRequestTimeout(); + this.timeout = opts.requestTimeout(); this.promise = promise; fsmRegistry.put(syncId, this); } @@ -60,7 +61,6 @@ public void start() { handlers.getOnBeforeSend().accept(request); } - // TODO: remove string format this.timer = runAfter( timeout, @@ -68,9 +68,7 @@ public void start() { if (handlers != null && handlers.getOnTimeout() != null) { handlers.getOnTimeout().accept(request); } - kill( - new TimeoutException( - String.format("Request timeout: %s; timeout = %sms", request, timeout))); + kill(new TimeoutException("Request timeout: " + request + "; timeout = " + timeout)); }); promise.whenComplete((iProtoResponse, throwable) -> this.timer.cancel()); @@ -119,12 +117,13 @@ public void kill(Throwable ex) { promise.completeExceptionally(ex); } - private Timeout runAfter(long after, Runnable cb) { - if (after == 0) { + private Timeout runAfter(Duration after, Runnable cb) { + if (after.isZero()) { cb.run(); return null; } - return timerService.newTimeout(timeoutHandler -> cb.run(), after, TimeUnit.MILLISECONDS); + return timerService.newTimeout( + timeoutHandler -> cb.run(), after.toNanos(), TimeUnit.NANOSECONDS); } } diff --git a/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java b/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java index 566d64cc..597aa4a7 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java +++ b/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java @@ -156,8 +156,8 @@ private void send() { future.completeExceptionally(new ClientException("watcher timeout")); } }, - opts.getSendTimeout(), - TimeUnit.MILLISECONDS); + opts.sendTimeout().toNanos(), + TimeUnit.NANOSECONDS); future.whenComplete( (r, exc) -> { timer.cancel(); diff --git a/tarantool-core/src/main/java/io/tarantool/core/protocol/requests/IProtoBegin.java b/tarantool-core/src/main/java/io/tarantool/core/protocol/requests/IProtoBegin.java index 76ce959c..57d73578 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/protocol/requests/IProtoBegin.java +++ b/tarantool-core/src/main/java/io/tarantool/core/protocol/requests/IProtoBegin.java @@ -6,6 +6,7 @@ package io.tarantool.core.protocol.requests; import java.io.IOException; +import java.time.Duration; import java.util.HashMap; import java.util.Map; @@ -22,18 +23,25 @@ import static io.tarantool.core.protocol.requests.IProtoConstant.RAW_IPROTO_TYPE_BEGIN; import static io.tarantool.core.protocol.requests.IProtoConstant.RAW_MAP_HEADER_WITH_ONE_ITEM; import static io.tarantool.core.protocol.requests.IProtoConstant.RAW_MAP_HEADER_WITH_TWO_ITEMS; +import io.tarantool.core.TimeoutsUtils; import io.tarantool.core.protocol.TransactionIsolationLevel; public class IProtoBegin extends IProtoBaseRequest { // body fields - private final long timeout; + private final Duration timeout; private final TransactionIsolationLevel level; - public IProtoBegin(Long streamId, long timeout, TransactionIsolationLevel level) { - if (timeout <= 0) { - throw new IllegalArgumentException("timeout should be greater than 0"); - } + /** + * @deprecated Use {@link #IProtoBegin(Long, Duration, TransactionIsolationLevel)} + */ + @Deprecated + public IProtoBegin(Long streamId, long timeoutMs, TransactionIsolationLevel level) { + this(streamId, Duration.ofMillis(timeoutMs), level); + } + + public IProtoBegin(Long streamId, Duration timeout, TransactionIsolationLevel level) { + TimeoutsUtils.checkTimeoutIsPositive(timeout); this.timeout = timeout; this.level = level; this.setStreamId(streamId); @@ -55,7 +63,9 @@ public byte[] getPacket(MessageBufferPacker packer) throws IOException { } packer.addPayload(RAW_IPROTO_TIMEOUT); // key - packer.packDouble(timeout); // value + + double totalSeconds = timeout.toNanos() / 1_000_000_000.0; + packer.packDouble(totalSeconds); // value if (!level.equals(TransactionIsolationLevel.DEFAULT)) { packer.addPayload(RAW_IPROTO_TXN_ISOLATION); // key @@ -68,7 +78,8 @@ public byte[] getPacket(MessageBufferPacker packer) throws IOException { @Override public MapValue getBody() { Map map = new HashMap<>(); - map.put(MP_IPROTO_TIMEOUT, ValueFactory.newInteger(timeout)); + double totalSeconds = timeout.toNanos() / 1_000_000_000.0; + map.put(MP_IPROTO_TIMEOUT, ValueFactory.newFloat(totalSeconds)); if (!level.equals(TransactionIsolationLevel.DEFAULT)) { map.put(MP_IPROTO_TXN_ISOLATION, ValueFactory.newInteger(level.ordinal())); diff --git a/tarantool-core/src/test/java/io/tarantool/core/integration/BaseTest.java b/tarantool-core/src/test/java/io/tarantool/core/integration/BaseTest.java index c4778f21..193e2d5b 100644 --- a/tarantool-core/src/test/java/io/tarantool/core/integration/BaseTest.java +++ b/tarantool-core/src/test/java/io/tarantool/core/integration/BaseTest.java @@ -7,6 +7,7 @@ import java.io.IOException; import java.net.InetSocketAddress; +import java.time.Duration; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -93,7 +94,7 @@ protected IProtoClient createClientAndConnect( null, null, useTupleExtension); - client.connect(address, 3_000).get(); + client.connect(address, Duration.ofSeconds(3)).toCompletableFuture().get(); return client; } } diff --git a/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionCloseOnClientSideTest.java b/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionCloseOnClientSideTest.java index fa284fff..fbea128d 100644 --- a/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionCloseOnClientSideTest.java +++ b/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionCloseOnClientSideTest.java @@ -6,6 +6,7 @@ package io.tarantool.core.integration; import java.net.InetSocketAddress; +import java.time.Duration; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; @@ -50,7 +51,7 @@ public void testConnectAndClose() throws Exception { closeFuture.completeExceptionally(ex); }); InetSocketAddress address = tt.mappedAddress(); - connection.connect(address, 3_000).get(); + connection.connect(address, Duration.ofSeconds(3)).toCompletableFuture().get(); Thread.sleep(500); connection.close(); Exception ex = assertThrows(CompletionException.class, closeFuture::join); @@ -70,7 +71,7 @@ public void testConnectAndCloseShutdown() throws Exception { closeFuture.completeExceptionally(ex); }); InetSocketAddress address = tt.mappedAddress(); - connection.connect(address, 3_000).get(); + connection.connect(address, Duration.ofSeconds(3)).toCompletableFuture().get(); Thread.sleep(500); connection.shutdownClose(); Exception ex = assertThrows(CompletionException.class, closeFuture::join); diff --git a/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionCloseOnServerSideTest.java b/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionCloseOnServerSideTest.java index f8e5d0b8..e4ddd06e 100644 --- a/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionCloseOnServerSideTest.java +++ b/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionCloseOnServerSideTest.java @@ -6,6 +6,7 @@ package io.tarantool.core.integration; import java.net.InetSocketAddress; +import java.time.Duration; import java.util.HashMap; import java.util.Map; import java.util.concurrent.CompletableFuture; @@ -73,7 +74,7 @@ public void testConnectAndCloseOnServer() throws Exception { flags.put(ConnectionCloseEvent.CLOSE_BY_SHUTDOWN, true); }); InetSocketAddress address = tt.mappedAddress(); - connection.connect(address, 3_000).get(); + connection.connect(address, Duration.ofSeconds(3)).toCompletableFuture().get(); tt.stop(); Exception ex = assertThrows(CompletionException.class, closeFuture::join); Throwable cause = ex.getCause(); diff --git a/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionToTarantoolTest.java b/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionToTarantoolTest.java index 2262cfbb..79085ed8 100644 --- a/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionToTarantoolTest.java +++ b/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionToTarantoolTest.java @@ -10,6 +10,7 @@ import java.net.InetSocketAddress; import java.net.SocketException; import java.nio.channels.ClosedChannelException; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; @@ -135,13 +136,14 @@ static void tearDown() { @Test public void testConnect() throws Exception { - Connection client = factory.create(); - InetSocketAddress address = tt.mappedAddress(); - CompletableFuture connectFuture = client.connect(address, 3_000); - Greeting greeting = connectFuture.get(); - assertEquals(greeting.getVersion(), version); - assertEquals(greeting.getProtocolType(), protocolType); - assertEquals(greeting.getInstanceUUID(), instanceUUID); + try (Connection client = factory.create()) { + InetSocketAddress address = tt.mappedAddress(); + final Greeting greeting = + client.connect(address, Duration.ofSeconds(3)).toCompletableFuture().join(); + assertEquals(greeting.getVersion(), version); + assertEquals(greeting.getProtocolType(), protocolType); + assertEquals(greeting.getInstanceUUID(), instanceUUID); + } } /* @@ -150,17 +152,19 @@ public void testConnect() throws Exception { * explaining, that tcp connect cannot be established */ @Test - public void testConnectToAddressWithBadPort() { - Connection client = factory.create(); - InetSocketAddress address = new InetSocketAddress(tt.getHost(), BAD_PORT); - CompletableFuture future = client.connect(address, 3_000); - Exception ex = assertThrows(CompletionException.class, future::join); - Throwable cause = ex.getCause(); - assertEquals(ConnectionException.class, cause.getClass()); - assertEquals( - String.format("Failed to connect to the Tarantool server at %s", address), - cause.getMessage()); - assertEquals(ConnectException.class, findRootCause(ex).getClass()); + public void testConnectToAddressWithBadPort() throws Exception { + try (Connection client = factory.create()) { + InetSocketAddress address = new InetSocketAddress(tt.getHost(), BAD_PORT); + CompletableFuture future = + client.connect(address, Duration.ofSeconds(3)).toCompletableFuture(); + Exception ex = assertThrows(CompletionException.class, future::join); + Throwable cause = ex.getCause(); + assertEquals(ConnectionException.class, cause.getClass()); + assertEquals( + String.format("Failed to connect to the Tarantool server at %s", address), + cause.getMessage()); + assertEquals(ConnectException.class, findRootCause(ex).getClass()); + } } /* @@ -170,22 +174,24 @@ public void testConnectToAddressWithBadPort() { * because all port numbers are valid in range 0..65535. */ @RepeatedTest(value = 3) - public void testConnectToAddressWithBadHost() { - Connection client = factory.create(); - InetSocketAddress address = new InetSocketAddress(BAD_HOST, tt.getFirstMappedPort()); - CompletableFuture future = client.connect(address, 5_000); - Exception ex = assertThrows(CompletionException.class, future::join); - Throwable cause = ex.getCause(); - if (cause.getClass() == ConnectionException.class) { - assertEquals( - String.format("Failed to connect to the Tarantool server at %s", address), - cause.getMessage()); - assertEquals(ConnectTimeoutException.class, findRootCause(ex).getClass()); - } else if (cause.getClass() == TimeoutException.class) { - assertEquals("Connection timeout", cause.getMessage()); - log.info(findRootCause(ex).getClass().toString()); // Need for debug in the future - } else { - fail("An unknown exception was thrown during connection to the bad hostname!", cause); + public void testConnectToAddressWithBadHost() throws Exception { + try (Connection client = factory.create()) { + InetSocketAddress address = new InetSocketAddress(BAD_HOST, tt.getFirstMappedPort()); + CompletableFuture future = + client.connect(address, Duration.ofSeconds(5)).toCompletableFuture(); + Exception ex = assertThrows(CompletionException.class, future::join); + Throwable cause = ex.getCause(); + if (cause.getClass() == ConnectionException.class) { + assertEquals( + String.format("Failed to connect to the Tarantool server at %s", address), + cause.getMessage()); + assertEquals(ConnectTimeoutException.class, findRootCause(ex).getClass()); + } else if (cause.getClass() == TimeoutException.class) { + assertEquals("Connection timeout", cause.getMessage()); + log.info(findRootCause(ex).getClass().toString()); // Need for debug in the future + } else { + fail("An unknown exception was thrown during connection to the bad hostname!", cause); + } } } @@ -196,48 +202,52 @@ public void testConnectToAddressWithBadHost() { * */ @Test @DisabledIfEnvironmentVariable(named = "TARANTOOL_VERSION", matches = "3.*") - public void testConnectToNonAcceptingService() { + public void testConnectToNonAcceptingService() throws Exception { Integer otherPort = tt.getMappedPort(3302); - Connection client = factory.create(); - InetSocketAddress address = new InetSocketAddress(tt.getHost(), otherPort); - CompletableFuture future = client.connect(address, 3_000); - Exception ex = assertThrows(CompletionException.class, future::join); - Throwable cause = findRootCause(ex); - - /* - * When we run container with disabled userland proxy we get - * SocketException. When userland proxy is enabled, we normally connect - * and then get closing immediately and ConnectionException is raised. - * */ - assertTrue( - Arrays.asList( - ConnectionException.class, - ConnectionClosedException.class, - IOException.class, - SocketException.class) - .contains(cause.getClass())); - String message = cause.getMessage(); - assertTrue( - Arrays.asList( - "Connection closed by server", - "Connection reset", - "Connection reset by peer", - "Соединение разорвано другой стороной") - .contains(message), - message); + try (Connection client = factory.create()) { + InetSocketAddress address = new InetSocketAddress(tt.getHost(), otherPort); + CompletableFuture future = + client.connect(address, Duration.ofSeconds(3)).toCompletableFuture(); + Exception ex = assertThrows(CompletionException.class, future::join); + Throwable cause = findRootCause(ex); + + /* + * When we run container with disabled userland proxy we get + * SocketException. When userland proxy is enabled, we normally connect + * and then get closing immediately and ConnectionException is raised. + * */ + assertTrue( + Arrays.asList( + ConnectionException.class, + ConnectionClosedException.class, + IOException.class, + SocketException.class) + .contains(cause.getClass())); + String message = cause.getMessage(); + assertTrue( + Arrays.asList( + "Connection closed by server", + "Connection reset", + "Connection reset by peer", + "Соединение разорвано другой стороной") + .contains(message), + message); + } } @Test @DisabledIfEnvironmentVariable(named = "TARANTOOL_VERSION", matches = "3.*") - public void testConnectToServiceWithBadGreeting() { + public void testConnectToServiceWithBadGreeting() throws Exception { Integer otherPort = tt.getMappedPort(3304); - Connection client = factory.create(); - InetSocketAddress address = new InetSocketAddress(tt.getHost(), otherPort); - CompletableFuture future = client.connect(address, 3_000); - Exception ex = assertThrows(CompletionException.class, future::join); - Throwable cause = findRootCause(ex); - assertEquals(BadGreetingException.class, cause.getClass()); - assertTrue(cause.getMessage().startsWith("bad greeting start:")); + try (Connection client = factory.create()) { + InetSocketAddress address = new InetSocketAddress(tt.getHost(), otherPort); + CompletableFuture future = + client.connect(address, Duration.ofSeconds(3)).toCompletableFuture(); + Exception ex = assertThrows(CompletionException.class, future::join); + Throwable cause = findRootCause(ex); + assertEquals(BadGreetingException.class, cause.getClass()); + assertTrue(cause.getMessage().startsWith("bad greeting start:")); + } } /* @@ -247,15 +257,17 @@ public void testConnectToServiceWithBadGreeting() { * */ @Test @DisabledIfEnvironmentVariable(named = "TARANTOOL_VERSION", matches = "3.*") - public void testConnectToSilentNode() { + public void testConnectToSilentNode() throws Exception { Integer otherPort = tt.getMappedPort(3303); - Connection client = factory.create().listen(msg -> {}); - InetSocketAddress address = new InetSocketAddress(tt.getHost(), otherPort); - CompletableFuture future = client.connect(address, 3000); - Exception ex = assertThrows(CompletionException.class, future::join); - Throwable cause = findRootCause(ex); - assertEquals(TimeoutException.class, cause.getClass()); - assertEquals("Connection timeout", cause.getMessage()); + try (Connection client = factory.create().listen(msg -> {})) { + InetSocketAddress address = new InetSocketAddress(tt.getHost(), otherPort); + CompletableFuture future = + client.connect(address, Duration.ofSeconds(3)).toCompletableFuture(); + Exception ex = assertThrows(CompletionException.class, future::join); + Throwable cause = findRootCause(ex); + assertEquals(TimeoutException.class, cause.getClass()); + assertEquals("Connection timeout", cause.getMessage()); + } } /* @@ -267,15 +279,17 @@ public void testConnectToSilentNode() { @DisabledIfEnvironmentVariable(named = "TARANTOOL_VERSION", matches = "3.*") public void testConnectToSilentNodeAndClose() throws Exception { Integer otherPort = tt.getMappedPort(3303); - Connection client = factory.create().listen(msg -> {}); - InetSocketAddress address = new InetSocketAddress(tt.getHost(), otherPort); - CompletableFuture future = client.connect(address, 3000); - Thread.sleep(100); - client.close(); - Exception ex = assertThrows(CompletionException.class, future::join); - Throwable cause = findRootCause(ex); - assertEquals(ConnectionClosedException.class, cause.getClass()); - assertEquals("Connection closed by client", cause.getMessage()); + try (Connection client = factory.create().listen(msg -> {})) { + InetSocketAddress address = new InetSocketAddress(tt.getHost(), otherPort); + CompletableFuture future = + client.connect(address, Duration.ofSeconds(3)).toCompletableFuture(); + Thread.sleep(100); + client.close(); + Exception ex = assertThrows(CompletionException.class, future::join); + Throwable cause = findRootCause(ex); + assertEquals(ConnectionClosedException.class, cause.getClass()); + assertEquals("Connection closed by client", cause.getMessage()); + } } /* @@ -286,16 +300,18 @@ public void testConnectToSilentNodeAndClose() throws Exception { */ @Test public void testConnectWithClosingOnClientSide() throws Exception { - Connection client = factory.create(); - InetSocketAddress address = new InetSocketAddress(BAD_HOST, BAD_PORT); - CompletableFuture future = client.connect(address, 3_000); - Thread.sleep(100); - client.close(); - Exception ex = assertThrows(CompletionException.class, future::join); - Throwable cause = ex.getCause(); - assertEquals(cause.getClass(), ConnectionException.class); - assertEquals("Connection closed by client", cause.getMessage()); - assertEquals(ClosedChannelException.class, findRootCause(ex).getClass()); + try (Connection client = factory.create()) { + InetSocketAddress address = new InetSocketAddress(BAD_HOST, BAD_PORT); + CompletableFuture future = + client.connect(address, Duration.ofSeconds(3)).toCompletableFuture(); + Thread.sleep(100); + client.close(); + Exception ex = assertThrows(CompletionException.class, future::join); + Throwable cause = ex.getCause(); + assertEquals(ConnectionException.class, cause.getClass()); + assertEquals("Connection closed by client", cause.getMessage()); + assertEquals(ClosedChannelException.class, findRootCause(ex).getClass()); + } } /** @@ -306,88 +322,101 @@ public void testConnectWithClosingOnClientSide() throws Exception { * @throws InterruptedException */ @Test - public void testConnectByMultipleThreads() throws InterruptedException { - Connection client = factory.create(); - InetSocketAddress address = tt.mappedAddress(); - LinkedBlockingQueue> promises = new LinkedBlockingQueue<>(); - for (int i = 0; i < CONCURRENT_THREADS_COUNT; i++) { - new Thread(() -> promises.add(client.connect(address, 3_000))).start(); - } - int failed = 0; - int success = 0; - for (int i = 0; i < CONCURRENT_THREADS_COUNT; i++) { - CompletableFuture p = promises.take(); - try { - p.get(); - success++; - } catch (ExecutionException ex) { - Throwable cause = ex.getCause(); - assertEquals(ConnectionException.class, cause.getClass()); - failed++; + public void testConnectByMultipleThreads() throws Exception { + try (Connection client = factory.create()) { + InetSocketAddress address = tt.mappedAddress(); + LinkedBlockingQueue> promises = new LinkedBlockingQueue<>(); + for (int i = 0; i < CONCURRENT_THREADS_COUNT; i++) { + new Thread( + () -> + promises.add( + client.connect(address, Duration.ofSeconds(3)).toCompletableFuture())) + .start(); + } + int failed = 0; + int success = 0; + for (int i = 0; i < CONCURRENT_THREADS_COUNT; i++) { + CompletableFuture p = promises.take(); + try { + p.get(); + success++; + } catch (ExecutionException ex) { + Throwable cause = ex.getCause(); + assertEquals(ConnectionException.class, cause.getClass()); + failed++; + } } + assertEquals(0, failed); + assertEquals(CONCURRENT_THREADS_COUNT, success); } - assertEquals(0, failed); - assertEquals(CONCURRENT_THREADS_COUNT, success); } @Test @DisabledIfEnvironmentVariable(named = "TARANTOOL_VERSION", matches = "3.*") - public void testConnectWithGreetingTimeout() throws InterruptedException { - Connection client = factory.create(); - InetSocketAddress address = new InetSocketAddress(tt.getHost(), tt.getMappedPort(3306)); - CompletableFuture future = client.connect(address, 1_000); - Exception ex = assertThrows(CompletionException.class, future::join); - Throwable cause = ex.getCause(); - assertTrue( - cause instanceof TimeoutException || cause instanceof ConnectionClosedException, - "Exception should be either TimeoutException or ConnectionClosedException"); - assertEquals("Connection timeout", cause.getMessage()); + public void testConnectWithGreetingTimeout() throws Exception { + try (Connection client = factory.create()) { + InetSocketAddress address = new InetSocketAddress(tt.getHost(), tt.getMappedPort(3306)); + CompletableFuture future = + client.connect(address, Duration.ofSeconds(1)).toCompletableFuture(); + Exception ex = assertThrows(CompletionException.class, future::join); + Throwable cause = ex.getCause(); + assertTrue( + cause instanceof TimeoutException || cause instanceof ConnectionClosedException, + "Exception should be either TimeoutException or ConnectionClosedException"); + assertEquals("Connection timeout", cause.getMessage()); + } } @Test @DisabledIfEnvironmentVariable(named = "TARANTOOL_VERSION", matches = "3.*") public void testConnectWithWaitingForGreetingAndClose() throws Exception { - Connection client = factory.create(); - InetSocketAddress address = new InetSocketAddress(tt.getHost(), tt.getMappedPort(3306)); - CompletableFuture future = client.connect(address, 2_000); - Thread.sleep(500); - client.close(); - Exception ex = assertThrows(CompletionException.class, future::join); - Throwable cause = ex.getCause(); - assertEquals(ConnectionClosedException.class, cause.getClass()); - assertEquals("Connection closed by client", cause.getMessage()); + try (Connection client = factory.create()) { + InetSocketAddress address = new InetSocketAddress(tt.getHost(), tt.getMappedPort(3306)); + CompletableFuture future = + client.connect(address, Duration.ofSeconds(2)).toCompletableFuture(); + Thread.sleep(500); + client.close(); + Exception ex = assertThrows(CompletionException.class, future::join); + Throwable cause = ex.getCause(); + assertEquals(ConnectionClosedException.class, cause.getClass()); + assertEquals("Connection closed by client", cause.getMessage()); + } } @Test - public void testConnectAndSendByMultipleThreads() throws InterruptedException { - Connection client = factory.create(); - InetSocketAddress address = tt.mappedAddress(); - LinkedBlockingQueue> promises = new LinkedBlockingQueue<>(); - IProtoRequest msg = createSelectRequest(0); - for (int i = 0; i < CONCURRENT_THREADS_COUNT; i++) { - new Thread( - () -> { - CompletableFuture future = - client.connect(address, 3_000).thenCompose(g -> client.send(msg)); - promises.add(future); - }) - .start(); - } - int failed = 0; - int success = 0; - for (int i = 0; i < CONCURRENT_THREADS_COUNT; i++) { - CompletableFuture p = promises.take(); - try { - p.get(); - success++; - } catch (ExecutionException ex) { - Throwable cause = ex.getCause(); - assertEquals(ConnectionException.class, cause.getClass()); - failed++; + public void testConnectAndSendByMultipleThreads() throws Exception { + try (Connection client = factory.create()) { + InetSocketAddress address = tt.mappedAddress(); + LinkedBlockingQueue> promises = new LinkedBlockingQueue<>(); + IProtoRequest msg = createSelectRequest(0); + for (int i = 0; i < CONCURRENT_THREADS_COUNT; i++) { + new Thread( + () -> { + CompletableFuture future = + client + .connect(address, Duration.ofSeconds(3)) + .thenCompose(g -> client.send(msg)) + .toCompletableFuture(); + promises.add(future); + }) + .start(); + } + int failed = 0; + int success = 0; + for (int i = 0; i < CONCURRENT_THREADS_COUNT; i++) { + CompletableFuture p = promises.take(); + try { + p.get(); + success++; + } catch (ExecutionException ex) { + Throwable cause = ex.getCause(); + assertEquals(ConnectionException.class, cause.getClass()); + failed++; + } } + assertEquals(0, failed); + assertEquals(CONCURRENT_THREADS_COUNT, success); } - assertEquals(0, failed); - assertEquals(CONCURRENT_THREADS_COUNT, success); } private IProtoRequest createSelectRequest(int syncId) { @@ -407,95 +436,98 @@ private IProtoRequest createSelectRequest(int syncId) { @Test public void testIProtoSendAndReceive() throws Exception { MessageConsumer consumer = new MessageConsumer(); - Connection client = factory.create().listen(consumer); - InetSocketAddress address = tt.mappedAddress(); - client.connect(address, 3_000).join(); - - Greeting greeting = client.getGreeting().get(); - IProtoRequest request = - new IProtoAuth(API_USER, CREDS.get(API_USER), greeting.getSalt(), AuthType.CHAP_SHA1); - request.setSyncId(0); - - client.send(request).join(); - - IProtoRequest msg = createSelectRequest(0); - assertNull(client.send(msg).join()); - Thread.sleep(100); - assertEquals(consumer.queue.size(), 2); - - consumer.queue.take(); - IProtoMessage reply = consumer.queue.take(); - Map header = reply.getHeader().asMapValue().map(); - assertEquals(newInteger(IPROTO_OK), header.get(newInteger(IPROTO_REQUEST_TYPE))); - assertEquals(newInteger(0), header.get(newInteger(IPROTO_SYNC_ID))); - - Map body = reply.getBody().asMapValue().map(); - // No data in space - assertEquals(newArray(), body.get(newInteger(IPROTO_DATA))); + + try (Connection client = factory.create().listen(consumer)) { + InetSocketAddress address = tt.mappedAddress(); + client.connect(address, Duration.ofSeconds(3)).toCompletableFuture().join(); + + Greeting greeting = client.getGreeting().get(); + IProtoRequest request = + new IProtoAuth(API_USER, CREDS.get(API_USER), greeting.getSalt(), AuthType.CHAP_SHA1); + request.setSyncId(0); + + client.send(request).join(); + + IProtoRequest msg = createSelectRequest(0); + assertNull(client.send(msg).join()); + Thread.sleep(100); + assertEquals(2, consumer.queue.size()); + + consumer.queue.take(); + IProtoMessage reply = consumer.queue.take(); + Map header = reply.getHeader().asMapValue().map(); + assertEquals(newInteger(IPROTO_OK), header.get(newInteger(IPROTO_REQUEST_TYPE))); + assertEquals(newInteger(0), header.get(newInteger(IPROTO_SYNC_ID))); + + Map body = reply.getBody().asMapValue().map(); + // No data in space + assertEquals(newArray(), body.get(newInteger(IPROTO_DATA))); + } } @Test public void testSendAndReceiveWithConcurrentClose() throws Exception { MessageConsumer consumer = new MessageConsumer(); - Connection client = factory.create().listen(consumer); - InetSocketAddress address = tt.mappedAddress(); - client.connect(address, 3_000).join(); + try (Connection client = factory.create().listen(consumer)) { + InetSocketAddress address = tt.mappedAddress(); + client.connect(address, Duration.ofSeconds(3)).toCompletableFuture().join(); - List> futures = new ArrayList<>(); - CountDownLatch errorsCounter = new CountDownLatch(5); - CountDownLatch totalRequests = new CountDownLatch(10); + List> futures = new ArrayList<>(); + CountDownLatch errorsCounter = new CountDownLatch(5); + CountDownLatch totalRequests = new CountDownLatch(10); - Thread sender = - new Thread( - () -> { - int syncId = 0; - while (errorsCounter.getCount() > 0) { - IProtoMessage msg = createSelectRequest(syncId++); - try { - CompletableFuture promise = client.send(msg); - futures.add(promise); - promise.whenComplete( - (r, e) -> { - if (e != null) { - errorsCounter.countDown(); - } - }); - totalRequests.countDown(); - } catch (IllegalStateException e) { - errorsCounter.countDown(); + Thread sender = + new Thread( + () -> { + int syncId = 0; + while (errorsCounter.getCount() > 0) { + IProtoMessage msg = createSelectRequest(syncId++); + try { + CompletableFuture promise = client.send(msg); + futures.add(promise); + promise.whenComplete( + (r, e) -> { + if (e != null) { + errorsCounter.countDown(); + } + }); + totalRequests.countDown(); + } catch (IllegalStateException e) { + errorsCounter.countDown(); + } } - } - }); - - sender.start(); - totalRequests.await(); - client.close(); - sender.join(); - - int connectionClosedByClient = 0; - int otherExceptions = 0; - int failedFutures = 0; - int successFutures = 0; - for (CompletableFuture future : futures) { - try { - future.join(); - successFutures++; - } catch (Exception e) { - failedFutures++; - assertEquals(CompletionException.class, e.getClass()); - Throwable cause = findRootCause(e); - assertEquals(ConnectionClosedException.class, cause.getClass()); - if (cause.getMessage().equals("Connection closed by client")) { - connectionClosedByClient++; - } else { - otherExceptions++; + }); + + sender.start(); + totalRequests.await(); + client.close(); + sender.join(); + + int connectionClosedByClient = 0; + int otherExceptions = 0; + int failedFutures = 0; + int successFutures = 0; + for (CompletableFuture future : futures) { + try { + future.join(); + successFutures++; + } catch (Exception e) { + failedFutures++; + assertEquals(CompletionException.class, e.getClass()); + Throwable cause = findRootCause(e); + assertEquals(ConnectionClosedException.class, cause.getClass()); + if (cause.getMessage().equals("Connection closed by client")) { + connectionClosedByClient++; + } else { + otherExceptions++; + } } } + assertTrue(failedFutures >= 0); + assertTrue(connectionClosedByClient >= 0); + assertTrue(successFutures >= 10); + assertEquals(0, otherExceptions); + assertEquals(failedFutures, connectionClosedByClient); } - assertTrue(failedFutures >= 0); - assertTrue(connectionClosedByClient >= 0); - assertTrue(successFutures >= 10); - assertEquals(0, otherExceptions); - assertEquals(failedFutures, connectionClosedByClient); } } diff --git a/tarantool-core/src/test/java/io/tarantool/core/integration/GracefulShutdownTest.java b/tarantool-core/src/test/java/io/tarantool/core/integration/GracefulShutdownTest.java index 38d774ec..b6a4156b 100644 --- a/tarantool-core/src/test/java/io/tarantool/core/integration/GracefulShutdownTest.java +++ b/tarantool-core/src/test/java/io/tarantool/core/integration/GracefulShutdownTest.java @@ -6,6 +6,7 @@ package io.tarantool.core.integration; import java.net.InetSocketAddress; +import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CompletableFuture; @@ -52,7 +53,7 @@ static void tearDown() { private IProtoClient getClientAndConnect() throws Exception { IProtoClient client = new IProtoClientImpl(factory, factory.getTimerService()); - client.connect(address, 3_000).get(); + client.connect(address, Duration.ofSeconds(3)).toCompletableFuture().get(); return client; } diff --git a/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientTest.java b/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientTest.java index f6ca2476..f85a2c62 100644 --- a/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientTest.java +++ b/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientTest.java @@ -7,6 +7,7 @@ import java.io.IOException; import java.net.InetSocketAddress; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -82,7 +83,7 @@ public class IProtoClientTest extends BaseTest { private static final IProtoRequestOpts DEFAULT_REQUEST_OPTS = - IProtoRequestOpts.empty().withRequestTimeout(5000); + IProtoRequestOpts.empty().withRequestTimeout(Duration.ofSeconds(5)); private static TarantoolContainer tt; private static int spaceAId; private static int spaceBId; @@ -1526,7 +1527,7 @@ public void testLocalTimerHasNoPending() throws Exception { null, null, true); - client.connect(address, 3_000).get(); + client.connect(address, Duration.ofSeconds(3)).toCompletableFuture().get(); client.authorize(API_USER, CREDS.get(API_USER)).join(); IProtoMessage message = client.ping().get(); checkMessageHeader(message, IPROTO_OK, 4); @@ -1556,7 +1557,7 @@ public void testId() throws Exception { ex = assertThrows(CompletionException.class, client::getServerProtocolVersion).getCause(); assertEquals("Call connect before getting server details", ex.getMessage()); - client.connect(address, 3_000).get(); + client.connect(address, Duration.ofSeconds(3)).toCompletableFuture().get(); client.authorize(API_USER, CREDS.get(API_USER)).join(); if (tarantoolVersion != '3') { iprotoFeature = EnumSet.range(STREAMS, PAGINATION); @@ -1687,7 +1688,7 @@ public void testCallTimeout() throws Exception { client.authorize(API_USER, CREDS.get(API_USER)).join(); ArrayValue args = ValueFactory.newArray(ValueFactory.newString("one"), ValueFactory.newInteger(1)); - IProtoRequestOpts opts = IProtoRequestOpts.empty().withRequestTimeout(1000); + IProtoRequestOpts opts = IProtoRequestOpts.empty().withRequestTimeout(Duration.ofSeconds(1)); CompletableFuture future = client.call("slow_echo", args, opts); Exception ex = assertThrows(CompletionException.class, future::join); Throwable cause = ex.getCause(); @@ -1707,7 +1708,7 @@ public void testCallTimeoutWithIgnoredPacketsHandler() throws Exception { client.onIgnoredPacket(packet -> ignoredPackets.put(packet.getSyncId(), packet)); ArrayValue args = ValueFactory.newArray(ValueFactory.newString("one"), ValueFactory.newInteger(1)); - IProtoRequestOpts opts = IProtoRequestOpts.empty().withRequestTimeout(1000); + IProtoRequestOpts opts = IProtoRequestOpts.empty().withRequestTimeout(Duration.ofSeconds(1)); CompletableFuture future = client.call("slow_echo", args, opts); Exception ex = assertThrows(CompletionException.class, future::join); Throwable cause = ex.getCause(); @@ -1732,7 +1733,7 @@ public void testRawCallTimeout() throws Exception { ArrayValue args = ValueFactory.newArray(ValueFactory.newString("one"), ValueFactory.newInteger(1)); byte[] rawArgs = ArrayValueToBytes(args); - IProtoRequestOpts opts = IProtoRequestOpts.empty().withRequestTimeout(1000); + IProtoRequestOpts opts = IProtoRequestOpts.empty().withRequestTimeout(Duration.ofSeconds(1)); CompletableFuture future = client.call("slow_echo", rawArgs, null, opts); Exception ex = assertThrows(CompletionException.class, future::join); Throwable cause = ex.getCause(); @@ -1745,7 +1746,7 @@ public void testCallNoTimeout() throws Exception { client.authorize(API_USER, CREDS.get(API_USER)).join(); ArrayValue args = ValueFactory.newArray(ValueFactory.newString("one"), ValueFactory.newInteger(1)); - IProtoRequestOpts opts = IProtoRequestOpts.empty().withRequestTimeout(1000); + IProtoRequestOpts opts = IProtoRequestOpts.empty().withRequestTimeout(Duration.ofSeconds(1)); IProtoMessage message = client.call("nonslow_echo", args, opts).get(); checkMessageHeader(message, IPROTO_OK, 4); assertEquals(args, message.getBodyArrayValue(IPROTO_DATA)); @@ -1758,7 +1759,7 @@ public void testRawCallNoTimeout() throws Exception { ArrayValue args = ValueFactory.newArray(ValueFactory.newString("one"), ValueFactory.newInteger(1)); byte[] rawArgs = ArrayValueToBytes(args); - IProtoRequestOpts opts = IProtoRequestOpts.empty().withRequestTimeout(1000); + IProtoRequestOpts opts = IProtoRequestOpts.empty().withRequestTimeout(Duration.ofSeconds(1)); IProtoMessage message = client.call("nonslow_echo", rawArgs, null, opts).get(); checkMessageHeader(message, IPROTO_OK, 4); assertEquals(args, message.getBodyArrayValue(IPROTO_DATA)); diff --git a/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientWatchersTest.java b/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientWatchersTest.java index e4c9aa3e..9f693f66 100644 --- a/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientWatchersTest.java +++ b/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientWatchersTest.java @@ -6,6 +6,7 @@ package io.tarantool.core.integration; import java.net.InetSocketAddress; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -57,7 +58,7 @@ static void tearDown() { private IProtoClient getClientAndConnect(TarantoolContainer tt) throws Exception { InetSocketAddress address = tt.mappedAddress(); IProtoClient client = new IProtoClientImpl(factory, factory.getTimerService()); - client.connect(address, 3_000).get(); + client.connect(address, Duration.ofSeconds(3)).toCompletableFuture().get(); return client; } @@ -181,7 +182,7 @@ private void testWatcherRecoveryAfterReconnectOnContainer( client.close(); Thread.sleep(100); InetSocketAddress address = tt.mappedAddress(); - client.connect(address, 3_000).get(); + client.connect(address, Duration.ofSeconds(3)).toCompletableFuture().get(); Thread.sleep(1000); TarantoolContainerClientHelper.executeCommand( diff --git a/tarantool-core/src/test/java/io/tarantool/core/integration/MVCCStreamsTest.java b/tarantool-core/src/test/java/io/tarantool/core/integration/MVCCStreamsTest.java index 65a6f688..08340a56 100644 --- a/tarantool-core/src/test/java/io/tarantool/core/integration/MVCCStreamsTest.java +++ b/tarantool-core/src/test/java/io/tarantool/core/integration/MVCCStreamsTest.java @@ -8,6 +8,7 @@ import java.net.InetSocketAddress; import java.nio.file.Path; import java.nio.file.Paths; +import java.time.Duration; import java.util.List; import java.util.Objects; import java.util.concurrent.CompletableFuture; @@ -133,7 +134,7 @@ private void checkNoTuple(String ttCheck) throws Exception { private IProtoClient getClientAndConnect() throws Exception { IProtoClient client = new IProtoClientImpl(factory, factory.getTimerService()); - client.connect(address, 3_000).get(); + client.connect(address, Duration.ofSeconds(3)).toCompletableFuture().get(); client.authorize(API_USER, CREDS.get(API_USER)).join(); return client; } @@ -143,7 +144,10 @@ public void testStreamBeginAndCommit() throws Exception { IProtoClient client = getClientAndConnect(); Long streamId = client.allocateStreamId(); IProtoMessage messageBegin = - client.begin(streamId, 3_000L, TransactionIsolationLevel.DEFAULT).get(); + client + .begin(streamId, Duration.ofSeconds(3), TransactionIsolationLevel.DEFAULT) + .toCompletableFuture() + .get(); IProtoRequestOpts opts = IProtoRequestOpts.empty().withStreamId(streamId); IProtoMessage messageInsertA = client.insert(spaceAId, null, tupleA, opts).get(); IProtoMessage messageInsertB = client.insert(spaceBId, null, tupleB, opts).get(); @@ -178,7 +182,10 @@ public void testStreamBeginAndRollback() throws Exception { IProtoClient client = getClientAndConnect(); Long streamId = client.allocateStreamId(); IProtoMessage messageBegin = - client.begin(streamId, 3_000L, TransactionIsolationLevel.DEFAULT).get(); + client + .begin(streamId, Duration.ofSeconds(3), TransactionIsolationLevel.DEFAULT) + .toCompletableFuture() + .get(); IProtoRequestOpts opts = IProtoRequestOpts.empty().withStreamId(streamId); IProtoMessage messageInsertA = client.insert(spaceAId, null, tupleA, opts).get(); IProtoMessage messageInsertB = client.insert(spaceBId, null, tupleB, opts).get(); @@ -196,7 +203,10 @@ public void testStreamBeginAndCommitWithYieldingCall() throws Exception { IProtoClient client = getClientAndConnect(); Long streamId = client.allocateStreamId(); IProtoMessage messageBegin = - client.begin(streamId, 3_000L, TransactionIsolationLevel.DEFAULT).get(); + client + .begin(streamId, Duration.ofSeconds(3), TransactionIsolationLevel.DEFAULT) + .toCompletableFuture() + .get(); IProtoRequestOpts opts = IProtoRequestOpts.empty().withStreamId(streamId); IProtoMessage messageInsertA = client.insert(spaceAId, null, tupleA, opts).get(); IProtoMessage messageCall = client.call("insert_c", ValueFactory.emptyArray(), opts).get(); @@ -217,7 +227,10 @@ public void testStreamBeginAndCommitWithThrowingCall() throws Exception { IProtoClient client = getClientAndConnect(); Long streamId = client.allocateStreamId(); IProtoMessage messageBegin = - client.begin(streamId, 3_000L, TransactionIsolationLevel.DEFAULT).get(); + client + .begin(streamId, Duration.ofSeconds(3), TransactionIsolationLevel.DEFAULT) + .toCompletableFuture() + .get(); IProtoRequestOpts opts = IProtoRequestOpts.empty().withStreamId(streamId); IProtoMessage messageInsertA = client.insert(spaceAId, null, tupleA, opts).get(); @@ -248,7 +261,11 @@ public void testStreamBeginAndCommitWithThrowingCall() throws Exception { public void testStreamAutoRollback() throws Exception { IProtoClient client = getClientAndConnect(); Long streamId = client.allocateStreamId(); - IProtoMessage messageBegin = client.begin(streamId, 1, TransactionIsolationLevel.DEFAULT).get(); + IProtoMessage messageBegin = + client + .begin(streamId, Duration.ofMillis(1), TransactionIsolationLevel.DEFAULT) + .toCompletableFuture() + .get(); Thread.sleep(3_000L); IProtoRequestOpts opts = IProtoRequestOpts.empty().withStreamId(streamId); diff --git a/tarantool-core/src/test/java/io/tarantool/core/protocol/fsm/WatcherStateMachineTest.java b/tarantool-core/src/test/java/io/tarantool/core/protocol/fsm/WatcherStateMachineTest.java index 3eae7827..b7b191d0 100644 --- a/tarantool-core/src/test/java/io/tarantool/core/protocol/fsm/WatcherStateMachineTest.java +++ b/tarantool-core/src/test/java/io/tarantool/core/protocol/fsm/WatcherStateMachineTest.java @@ -6,10 +6,12 @@ package io.tarantool.core.protocol.fsm; import java.net.InetSocketAddress; +import java.time.Duration; import java.util.HashMap; import java.util.Map; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; import java.util.concurrent.TimeUnit; import java.util.function.BiConsumer; import java.util.function.Consumer; @@ -113,7 +115,7 @@ void registeredFailsWhenSendTimesOut() { 1L, response -> {}, connection, - WatcherOptions.builder().withSendTimeout(50).build(), + WatcherOptions.builder().withSendTimeout(Duration.ofMillis(50)).build(), timer); watcher.runOnce(); @@ -140,6 +142,12 @@ public CompletableFuture connect(InetSocketAddress addr, long timeoutM return new CompletableFuture<>(); } + @Override + public CompletionStage connect(InetSocketAddress address, Duration timeout) + throws IllegalStateException { + return new CompletableFuture<>(); + } + @Override public Connection listen(Consumer listener) { return this; @@ -168,6 +176,9 @@ public void resume() {} @Override public void setIdleTimeout(int idleTimeout) {} + @Override + public void setIdleTimeout(Duration timeout) {} + @Override public boolean isPaused() { return false; diff --git a/tarantool-core/src/test/java/io/tarantool/core/unit/requests/RequestTest.java b/tarantool-core/src/test/java/io/tarantool/core/unit/requests/RequestTest.java index 70877034..ada8951d 100644 --- a/tarantool-core/src/test/java/io/tarantool/core/unit/requests/RequestTest.java +++ b/tarantool-core/src/test/java/io/tarantool/core/unit/requests/RequestTest.java @@ -6,6 +6,7 @@ package io.tarantool.core.unit.requests; import java.nio.charset.StandardCharsets; +import java.time.Duration; import java.util.Arrays; import java.util.List; @@ -60,7 +61,7 @@ public void testRequestsToString() { // begin final Long streamId = 1L; - long timeout = 1L; + Duration timeout = Duration.ofSeconds(1L); IProtoRequest begin = new IProtoBegin(streamId, timeout, TransactionIsolationLevel.DEFAULT); assertEquals( "IProtoBegin(syncId = " diff --git a/tarantool-jackson-mapping/src/test/java/io/tarantool/mapping/benchmark/CustomBenchmarkRunner.java b/tarantool-jackson-mapping/src/test/java/io/tarantool/mapping/benchmark/CustomBenchmarkRunner.java index 18b3d52f..5236de2c 100644 --- a/tarantool-jackson-mapping/src/test/java/io/tarantool/mapping/benchmark/CustomBenchmarkRunner.java +++ b/tarantool-jackson-mapping/src/test/java/io/tarantool/mapping/benchmark/CustomBenchmarkRunner.java @@ -6,6 +6,7 @@ package io.tarantool.mapping.benchmark; import java.net.InetSocketAddress; +import java.time.Duration; import java.time.Instant; import java.util.ArrayList; import java.util.Arrays; @@ -71,7 +72,10 @@ private static void createClients(Integer connections) clients = new ArrayList<>(); for (int i = 0; i < connections; i++) { IProtoClient client = new IProtoClientImpl(factory, factory.getTimerService()); - client.connect(new InetSocketAddress(host, port), 3_000).get(); + client + .connect(new InetSocketAddress(host, port), Duration.ofSeconds(3)) + .toCompletableFuture() + .get(); log.debug( "Successfully connected to Tarantool, version = {}", client.eval("return _TARANTOOL", ValueFactory.emptyArray()).get()); @@ -139,7 +143,7 @@ public static Boolean simpleCallAndJacksonMapping( "return return_one_tuple()", STUB_ARGS, null, - IProtoRequestOpts.empty().withRequestTimeout(2_000)) + IProtoRequestOpts.empty().withRequestTimeout(Duration.ofSeconds(2))) .thenApply(TarantoolJacksonMapping::readResponse)); } diff --git a/tarantool-jackson-mapping/src/test/java/io/tarantool/mapping/integration/IProtoClientTest.java b/tarantool-jackson-mapping/src/test/java/io/tarantool/mapping/integration/IProtoClientTest.java index d62ce309..1479f916 100644 --- a/tarantool-jackson-mapping/src/test/java/io/tarantool/mapping/integration/IProtoClientTest.java +++ b/tarantool-jackson-mapping/src/test/java/io/tarantool/mapping/integration/IProtoClientTest.java @@ -9,6 +9,7 @@ import java.math.BigInteger; import java.math.RoundingMode; import java.net.InetSocketAddress; +import java.time.Duration; import java.time.Instant; import java.time.LocalDate; import java.time.LocalDateTime; @@ -1111,7 +1112,7 @@ public void testMapWithNonStringKeys(T object, TypeReference> expect private IProtoClient getClientAndConnect() throws Exception { IProtoClient client = new IProtoClientImpl(factory, factory.getTimerService()); - client.connect(address, 3_000).get(); + client.connect(address, Duration.ofSeconds(3)).toCompletableFuture().get(); client.authorize(API_USER, CREDS.get(API_USER)).join(); return client; } diff --git a/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java b/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java index dfc4d175..0e4b65fe 100644 --- a/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java +++ b/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java @@ -5,8 +5,10 @@ package io.tarantool.pool; +import java.time.Duration; import java.util.ArrayDeque; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; import java.util.concurrent.TimeUnit; import java.util.function.BiFunction; import java.util.function.Consumer; @@ -48,7 +50,7 @@ final class PoolEntry { * @see io.tarantool.core.protocol.IProtoRequestOpts */ private static final IProtoRequestOpts firstPingOpts = - IProtoRequestOpts.empty().withRequestTimeout(1000); + IProtoRequestOpts.empty().withRequestTimeout(Duration.ofSeconds(1)); /** Logger instance. */ private static final Logger log = LoggerFactory.getLogger(PoolEntry.class); @@ -131,7 +133,7 @@ final class PoolEntry { *

It will be returned to all out clients wanting to obtain this client. It is recreated only * if client is reconnected. */ - private CompletableFuture connectFuture; + private CompletionStage connectFuture; /** Last heartbeat state/event. */ private HeartbeatEvent lastHeartbeatEvent; @@ -249,7 +251,8 @@ public PoolEntry( } else { this.heartbeatOpts = heartbeatOpts; this.heartbeatPingOpts = - IProtoRequestOpts.empty().withRequestTimeout(heartbeatOpts.getPingInterval()); + IProtoRequestOpts.empty() + .withRequestTimeout(Duration.ofMillis(heartbeatOpts.getPingInterval())); this.deathThreshold = heartbeatOpts.getDeathThreshold(); this.failedPingsThreshold = heartbeatOpts.getInvalidationThreshold(); this.windowSize = heartbeatOpts.getWindowSize(); @@ -345,7 +348,7 @@ public void shutdown() { */ public synchronized CompletableFuture connect() { if (connectFuture != null) { - return connectFuture; + return connectFuture.toCompletableFuture(); } return internalConnect(); } @@ -401,8 +404,8 @@ public void stopHeartbeat() { private CompletableFuture internalConnect() { log.info("connect {}/{}", tag, index); LongTaskTimer.Sample timer = startTimer(connectTime); - CompletableFuture future = - client.connect(group.getAddress(), connectTimeout, gracefulShutdown); + CompletionStage future = + client.connect(group.getAddress(), Duration.ofMillis(connectTimeout), gracefulShutdown); String user = group.getUser(); connectFuture = future @@ -417,7 +420,7 @@ private CompletableFuture internalConnect() { .thenCompose(r -> client.awaitConnectionReady()) .thenApply(r -> client) .whenComplete(this::onConnectComplete); - return connectFuture; + return connectFuture.toCompletableFuture(); } /** diff --git a/tarantool-pooling/src/test/java/io/tarantool/pool/integration/ConnectionPoolTest.java b/tarantool-pooling/src/test/java/io/tarantool/pool/integration/ConnectionPoolTest.java index b98015b5..2830bba1 100644 --- a/tarantool-pooling/src/test/java/io/tarantool/pool/integration/ConnectionPoolTest.java +++ b/tarantool-pooling/src/test/java/io/tarantool/pool/integration/ConnectionPoolTest.java @@ -5,6 +5,7 @@ package io.tarantool.pool.integration; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -584,7 +585,7 @@ public void testRequestsWithIgnoredPacketsHandler() throws Exception { .withTag("node-b") .build())); - IProtoRequestOpts opts = IProtoRequestOpts.empty().withRequestTimeout(1000); + IProtoRequestOpts opts = IProtoRequestOpts.empty().withRequestTimeout(Duration.ofSeconds(1)); ArrayValue args = ValueFactory.newArray(ValueFactory.newString("one"), ValueFactory.newInteger(1)); List> futures = new ArrayList<>(); diff --git a/tarantool-schema/src/test/java/io/tarantool/client/integration/TarantoolSchemaFetcherTest.java b/tarantool-schema/src/test/java/io/tarantool/client/integration/TarantoolSchemaFetcherTest.java index 55d18778..a7c0e558 100644 --- a/tarantool-schema/src/test/java/io/tarantool/client/integration/TarantoolSchemaFetcherTest.java +++ b/tarantool-schema/src/test/java/io/tarantool/client/integration/TarantoolSchemaFetcherTest.java @@ -6,6 +6,7 @@ package io.tarantool.client.integration; import java.net.InetSocketAddress; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -100,7 +101,11 @@ public static void setUp() { TarantoolContainerClientHelper.execInitScript(tt); client = new IProtoClientImpl(factory, timerService); - client.connect(new InetSocketAddress(tt.getHost(), tt.getFirstMappedPort()), 3_000).join(); + client + .connect( + new InetSocketAddress(tt.getHost(), tt.getFirstMappedPort()), Duration.ofSeconds(3)) + .toCompletableFuture() + .join(); client.authorize(API_USER, CREDS.get(API_USER)).join(); }