diff --git a/CHANGELOG.md b/CHANGELOG.md index 23db58fa..de4ff77c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,8 @@ method would return `null` instead of a `CompletableFuture`. The method now always returns a `CompletableFuture` object, which either completes successfully (upon receipt of a greeting) or terminates with an error. Providing not positive timeout value throws an exception. +- Fixed connection close callbacks executing before the connection state becomes closed. +- Fixed greeting object remaining non-null after connection closure. ### Documentation 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..4d46b975 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 @@ -14,28 +14,118 @@ import io.tarantool.core.protocol.IProtoMessage; import io.tarantool.core.protocol.IProtoResponse; +/** + * Tarantool node connection interface. + * + * @implSpec if an implementation of this interface represents a connection that allows reconnecting + * via the same {@link Connection} object, the implementation must document this behavior. + */ public interface Connection extends AutoCloseable { + /** + * Connects to a Tarantool node asynchronously. + * + * @param addr the Tarantool node address, represented as an {@link InetSocketAddress} object. + * Must not be null. + * @param timeoutMs the time from the start of the connection attempt until the actual connection + * is established. If the connection to the Tarantool node is not established after this time + * expires, the connection transitions to an unconnected state, and the {@link + * CompletableFuture} promise completes exceptionally. + * @return a connection promise represented as a {@link CompletableFuture} object. Completes + * successfully when the connection to the node is actually established, or exceptionally if + * the connection fails for any reason. Never null. + * @throws IllegalStateException if a connection attempt is made in a state where connecting is + * prohibited. The states in which connecting is prohibited are defined by the implementation. + * @throws IllegalArgumentException if {@code timeoutMs <= 0}. + */ CompletableFuture connect(InetSocketAddress addr, long timeoutMs) throws IllegalStateException; + /** + * Sends a message to a Tarantool node asynchronously. + * + * @param message the message, represented as an {@link IProtoMessage} object. Must not be null. + * @return a {@link CompletableFuture} object representing the message delivery result. Completes + * successfully if the message is sent to the Tarantool node, or exceptionally if the message + * fails to send. Never null. + * @throws IllegalStateException if, at the time of sending, the connection is not in a + * ready-to-send state. + */ CompletableFuture send(IProtoMessage message) throws IllegalStateException; + /** + * Registers a callback for handling any messages from a Tarantool node, which are represented as + * an {@link IProtoMessage} object. + * + * @param listener the callback for handling any messages from the Tarantool node. Must not be + * null. + * @return the same {@link Connection} object on which this method was invoked. + * @implSpec any implementation-specific details of this method - such as overwriting the previous + * callback upon multiple invocations, or other distinct behaviors - must be documented in the + * method's implementation documentation. + */ Connection listen(Consumer listener); + /** + * Registers a callback to receive notifications when the connection is closed for any reason, as + * reflected by a {@link ConnectionCloseEvent}. + * + * @param event the event that caused the connection to close. + * @param handler the connection close event handler. + * @return the same connection object on which this method was invoked. + * @apiNote when the connection is closed for any reason, only the callbacks that were subscribed + * to the corresponding event will be invoked. + * @implSpec implementations must guarantee that the connection is in a closed state at the moment + * the callback is invoked. + */ Connection onClose(ConnectionCloseEvent event, BiConsumer handler); + /** + * Returns the {@link Greeting} object. + * + * @return {@link Optional#empty()} if the connection is not in a connected state. + */ Optional getGreeting(); + /** + * Returns whether the connection is in a connected state at the time of invocation. + * + * @return {@code true} if the connection is in a connected state at the moment this method + * executes. + */ boolean isConnected(); + /** + * Initiates connection closure of the connection asynchronously, reporting the close cause as a + * graceful shutdown of the server. + * + *

This method is intended to be called upon receiving the {@code box.shutdown} notification + * from Tarantool. At this moment, the connection is still operational, but the server is + * preparing to stop. + * + *

In contrast to {@link #close()}, the close is reported to listeners as {@link + * ConnectionCloseEvent#CLOSE_BY_SHUTDOWN} rather than {@link + * ConnectionCloseEvent#CLOSE_BY_CLIENT}. Their reaction to these events is expected to differ. + * + * @see Tarantool + * documentation + */ void shutdownClose(); + /** Initiates connection closure via a {@link ConnectionCloseEvent} asynchronously. */ + @Override + void close() throws Exception; + + /** Sets the idle timeout duration in milliseconds. */ + void setIdleTimeout(int idleTimeoutMs); + + /** Stops reading incoming data from the socket. */ void pause(); + /** Resumes reading incoming data previously stopped by {@link #pause()}. */ void resume(); - void setIdleTimeout(int idleTimeout); - + /** Returns whether reading of incoming data is currently paused. */ boolean isPaused(); } diff --git a/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionCloseEvent.java b/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionCloseEvent.java index b73b2ce3..27104858 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionCloseEvent.java +++ b/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionCloseEvent.java @@ -5,9 +5,33 @@ package io.tarantool.core.connection; +/** {@link Connection} close events. */ public enum ConnectionCloseEvent { + + /** + * This event reflects a situation where the connection is closed from the network side—for + * example, due to a network disconnection. + */ CLOSE_BY_REMOTE, + + /** + * An event reflecting a situation where the connection has been closed by the client-side calling + * code. + */ CLOSE_BY_CLIENT, + + /** + * This event reflects a situation where a connection is closed from the Tarantool side via a call + * to {@code box.shutdown}. Unlike {@link #CLOSE_BY_REMOTE}, this event does not indicate network + * issues; rather, it signifies that a message has been received from a Tarantool node initiating + * a graceful shutdown. + * + * @see Tarantool + * documentation + */ CLOSE_BY_SHUTDOWN, + + /** An event representing a situation where a connection has been closed due to a timeout */ CLOSE_BY_TIMEOUT; } 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 48825a2a..b3d31e40 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 @@ -62,7 +62,7 @@ public String getMessage() { } } - private enum State { + public enum State { /** Closed state. */ CLOSED(ERR_NOT_CONNECTED), /** Connecting state. */ @@ -371,14 +371,21 @@ private synchronized void onChannelClose(ChannelFuture f) { message = ERR_CLOSED_BY_SERVER; event = ConnectionCloseEvent.CLOSE_BY_REMOTE; } - if (state.get() != State.CONNECTING) { - Exception ex = new ConnectionClosedException(message, f.cause()); + + final boolean notify = state.get() != State.CONNECTING; + final Exception ex = notify ? new ConnectionClosedException(message, f.cause()) : null; + if (notify) { failConnectPromise(ex); - fire(event, ex); } - state.set(State.CLOSED); + channel = null; connectPromise = null; + greeting = null; + state.set(State.CLOSED); + + if (notify) { + fire(event, ex); + } } private void completePromise(CompletableFuture promise, T result) { 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 f3aa6f1e..77ea1ee4 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 @@ -6,10 +6,12 @@ package io.tarantool.core.integration; import java.io.IOException; +import java.lang.reflect.Field; import java.net.ConnectException; 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; @@ -21,7 +23,9 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -58,6 +62,9 @@ import static io.tarantool.core.protocol.requests.IProtoConstant.IPROTO_SPACE_ID; import static io.tarantool.core.protocol.requests.IProtoConstant.IPROTO_SYNC_ID; import io.tarantool.core.connection.Connection; +import io.tarantool.core.connection.ConnectionCloseEvent; +import io.tarantool.core.connection.ConnectionImpl; +import io.tarantool.core.connection.ConnectionImpl.State; import io.tarantool.core.connection.Greeting; import io.tarantool.core.connection.exceptions.BadGreetingException; import io.tarantool.core.connection.exceptions.ConnectionClosedException; @@ -145,6 +152,65 @@ public void testConnect() throws Exception { assertEquals(greeting.getInstanceUUID(), instanceUUID); } + @Test + void testFireOnCloseCallbacksInClosedState() throws Throwable { + try (Connection connection = factory.create()) { + final CountDownLatch latch = new CountDownLatch(1); + final InetSocketAddress address = tt.mappedAddress(); + final AtomicReference ex = new AtomicReference<>(); + + connection.onClose( + ConnectionCloseEvent.CLOSE_BY_CLIENT, + (c, e) -> { + try { + final Field stateField = ConnectionImpl.class.getDeclaredField("state"); + stateField.setAccessible(true); + + Assertions.assertEquals( + State.CLOSED, ((AtomicReference) stateField.get(connection)).get()); + + } catch (Throwable exception) { + ex.set(exception); + } finally { + latch.countDown(); + } + }); + + connection.connect(address, Duration.ofSeconds(3).toMillis()).join(); + connection.close(); + + if (!latch.await(2, TimeUnit.MINUTES)) { + Assertions.fail("Test should complete"); + } + + if (ex.get() != null) { + throw ex.get(); + } + } + } + + @Test + void testGreetingNullAfterClosing() throws Exception { + try (Connection connection = factory.create()) { + final InetSocketAddress address = tt.mappedAddress(); + final CountDownLatch latch = new CountDownLatch(1); + connection.onClose(ConnectionCloseEvent.CLOSE_BY_CLIENT, (c, e) -> latch.countDown()); + + Assertions.assertFalse(connection.getGreeting().isPresent(), "Must be empty"); + connection.connect(address, Duration.ofSeconds(3).toMillis()).join(); + + Assertions.assertTrue(connection.getGreeting().isPresent(), "Must not be empty"); + + connection.close(); + + if (!latch.await(2, TimeUnit.MINUTES)) { + Assertions.fail("Connection must be disconnected"); + } + + Assertions.assertFalse(connection.getGreeting().isPresent(), "Must be empty"); + } + } + /* * Connect to resolvable host and bad port. In this case ConnectionException is * being thrown, and its root cause should be ConnectException with reason,