Merge pull request #402 from project-tsurugi/wip/i_601_apply_comments

apply commnets
project-tsurugi · Jan 25, 2024 · ea2ba04 · ea2ba04
2 parents 9404624 + 6fffff5
commit ea2ba04
Show file tree

Hide file tree

Showing 4 changed files with 50 additions and 19 deletions.
diff --git a/...ommon/src/main/java/com/tsurugidb/tsubakuro/channel/common/connection/wire/impl/Link.java b/...ommon/src/main/java/com/tsurugidb/tsubakuro/channel/common/connection/wire/impl/Link.java
@@ -19,8 +19,8 @@ public abstract class Link implements ServerResource {
     private long receivedMessageNumber = 0;
 
     protected ResponseBox responseBox = new ResponseBox(this);
-    protected TimeUnit timeUnit;
-    protected long timeout = 0;
+    protected TimeUnit closeTimeUnit;
+    protected long closeTimeout = 0;
 
     /**
      * Getter of the receivedMessageNumber.
@@ -114,8 +114,8 @@ public void pullMessage(long checkedMessageNumber, long t, TimeUnit u) throws Ti
      * @param t the timeout
      */
     public void setCloseTimeout(Timeout t) {
-        timeout = t.value();
-        timeUnit = t.unit();
+        closeTimeout = t.value();
+        closeTimeUnit = t.unit();
     }
 
     /**
@@ -140,9 +140,9 @@ ResponseBox getResponseBox() {
 
     // to suppress spotbug error
     long value() {
-        return this.timeout;
+        return this.closeTimeout;
     }
     TimeUnit unit() {
-        return this.timeUnit;
+        return this.closeTimeUnit;
     }
 }
diff --git a/...n/src/main/java/com/tsurugidb/tsubakuro/channel/common/connection/wire/impl/WireImpl.java b/...n/src/main/java/com/tsurugidb/tsubakuro/channel/common/connection/wire/impl/WireImpl.java
@@ -143,15 +143,21 @@ public void setCloseTimeout(Timeout timeout) {
      */
     @Override
     public void close() throws IOException {
-        try {
-            if (!closed.get()) {
+        if (!closed.getAndSet(true)) {
+            try {
                 link.close();
-                closed.set(true);
+            } catch (ServerException | InterruptedException e) {
+                throw new IOException(e);
             }
-        } catch (ServerException | InterruptedException e) {
-            throw new IOException(e);
         }
+    }
 
+    /**
+     * Set wire to close state without link close.
+     * The link must be closed for this method to be called.
+     */
+    public void closeWithoutGet() {
+        closed.set(true);
     }
 
     private static EndpointRequest.Request.Builder newRequest() {

diff --git a/modules/stream/src/main/java/com/tsurugidb/tsubakuro/channel/stream/StreamLink.java b/modules/stream/src/main/java/com/tsurugidb/tsubakuro/channel/stream/StreamLink.java
@@ -150,11 +150,24 @@ private boolean doPull(long timeout, TimeUnit unit, boolean throwException) thro
         }
     }
 
+    private int closeTimeoutMillis() {
+        if (closeTimeout == 0) {
+            return 0;
+        }
+        if (closeTimeUnit.toMillis(closeTimeout) > Integer.MAX_VALUE) {
+            return Integer.MAX_VALUE;
+         }
+         return (int) closeTimeUnit.toMillis(closeTimeout);
+    }
+
     private void closeBoxes(boolean intentionalClose) throws IOException {
         responseBox.doClose(intentionalClose);
         resultSetBox.doClose(intentionalClose);
+        if (!socket.isClosed()) {
+            socket.setSoTimeout(closeTimeoutMillis());
+            socket.close();
+        }
         socketClosed.set(true);
-        socket.close();
     }
 
     public ResultSetBox getResultSetBox() {
@@ -281,7 +294,7 @@ public void close() throws IOException, ServerException {
             try (var c1 = socket; var c2 = inStream; var c3 = outStream) {
                 send(REQUEST_SESSION_BYE, TERMINATION_REQUEST);
                 while (!socketClosed.get()) {
-                    doPull(timeout, timeUnit);
+                    doPull(closeTimeout, closeTimeUnit);
                 }
             } catch (TimeoutException e) {
                 socketError.set(true);
@@ -295,10 +308,12 @@ public void close() throws IOException, ServerException {
         }
     }
 
-    public void closeWithoutGet() throws IOException, ServerException {
+    /**
+     * Close the socket without sending REQUEST_SESSION_BYE.
+     * This method is intended to use before session open.
+     */
+    public void closeWithoutGet() throws IOException {
         closed.set(true);
-        socketClosed.set(true);
-        socket.close();
         closeBoxes(false);
     }
 }
diff --git a/...src/main/java/com/tsurugidb/tsubakuro/channel/stream/connection/FutureStreamWireImpl.java b/...src/main/java/com/tsurugidb/tsubakuro/channel/stream/connection/FutureStreamWireImpl.java
@@ -34,7 +34,12 @@ public Wire get() throws IOException, ServerException, InterruptedException {
                 wireImpl.setSessionID(futureSessionID.get());
                 return wireImpl;
             } catch (IOException | ServerException | InterruptedException e) {
-                streamLink.closeWithoutGet();
+                try {
+                    streamLink.closeWithoutGet();
+                    wireImpl.closeWithoutGet();
+                } catch (IOException exceptionOnClose) {
+                    e.addSuppressed(exceptionOnClose);
+                }
                 throw e;
             }
         }
@@ -48,7 +53,12 @@ public Wire get(long timeout, TimeUnit unit) throws IOException, ServerException
                 wireImpl.setSessionID(futureSessionID.get(timeout, unit));
                 return wireImpl;
             } catch (IOException | ServerException | InterruptedException | TimeoutException e) {
-                streamLink.closeWithoutGet();
+                try {
+                    streamLink.closeWithoutGet();
+                    wireImpl.closeWithoutGet();
+                } catch (IOException exceptionOnClose) {
+                    e.addSuppressed(exceptionOnClose);
+                }
                 throw e;
             }
         }
@@ -66,7 +76,7 @@ public void close() throws IOException, ServerException, InterruptedException {
         if (!gotton.getAndSet(true)) {
             futureSessionID.get();  // ensure notify client of session limit error 
             streamLink.closeWithoutGet();
-            wireImpl.close();
+            wireImpl.closeWithoutGet();
         }
     }
 }