[ISSUE #5095] Add some remoting test (#5251)

* [ISSUE #5095] Add some remoting test * [ISSUE #5095] Add some remoting test * [ISSUE #5095] Add some remoting test Co-authored-by: 斜阳 <terrance.lzm@alibaba-inc.com>
apache · Feb 21, 2023 · 7342fe7 · 7342fe7
1 parent b253658
commit 7342fe7
Show file tree

Hide file tree

Showing 2 changed files with 41 additions and 5 deletions.
diff --git a/remoting/src/test/java/org/apache/rocketmq/remoting/TlsTest.java b/remoting/src/test/java/org/apache/rocketmq/remoting/TlsTest.java
@@ -23,10 +23,13 @@
 import java.io.FileWriter;
 import java.io.IOException;
 import java.io.PrintWriter;
+import java.net.Socket;
 import java.util.UUID;
 import java.io.InputStream;
 import java.io.BufferedInputStream;
 import java.io.BufferedOutputStream;
+import java.util.concurrent.TimeUnit;
+import org.apache.rocketmq.remoting.common.RemotingUtil;
 import org.apache.rocketmq.remoting.common.TlsMode;
 import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
 import org.apache.rocketmq.remoting.netty.NettyClientConfig;
@@ -70,6 +73,7 @@
 import static org.apache.rocketmq.remoting.netty.TlsSystemConfig.tlsTestModeEnable;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.failBecauseExceptionWasNotThrown;
+import static org.awaitility.Awaitility.await;
 import static org.junit.Assert.assertNotNull;
 
 @RunWith(MockitoJUnitRunner.class)
@@ -147,6 +151,8 @@ else if ("noClientAuthFailure".equals(name.getMethodName())) {
 
         remotingServer = RemotingServerTest.createRemotingServer();
         remotingClient = RemotingServerTest.createRemotingClient(clientConfig);
+
+        await().atMost(200, TimeUnit.MILLISECONDS).until(() -> isHostConnectable(getServerAddress()));
     }
 
     @After
@@ -203,7 +209,7 @@ public void serverAcceptsUnAuthClient() throws Exception {
     @Test
     public void serverRejectsSSLClient() throws Exception {
         try {
-            RemotingCommand response = remotingClient.invokeSync("localhost:" + remotingServer.localListenPort(), createRequest(), 1000 * 5);
+            RemotingCommand response = remotingClient.invokeSync(getServerAddress(), createRequest(), 1000 * 5);
             failBecauseExceptionWasNotThrown(RemotingSendRequestException.class);
         } catch (RemotingSendRequestException ignore) {
         }
@@ -216,7 +222,7 @@ public void serverRejectsSSLClient() throws Exception {
     @Test
     public void serverRejectsUntrustedClientCert() throws Exception {
         try {
-            RemotingCommand response = remotingClient.invokeSync("localhost:" + remotingServer.localListenPort(), createRequest(), 1000 * 5);
+            RemotingCommand response = remotingClient.invokeSync(getServerAddress(), createRequest(), 1000 * 5);
             failBecauseExceptionWasNotThrown(RemotingSendRequestException.class);
         } catch (RemotingSendRequestException ignore) {
         }
@@ -234,7 +240,7 @@ public void serverAcceptsUntrustedClientCert() throws Exception {
     @Test
     public void noClientAuthFailure() throws Exception {
         try {
-            RemotingCommand response = remotingClient.invokeSync("localhost:" + remotingServer.localListenPort(), createRequest(), 1000 * 3);
+            RemotingCommand response = remotingClient.invokeSync(getServerAddress(), createRequest(), 1000 * 3);
             failBecauseExceptionWasNotThrown(RemotingSendRequestException.class);
         } catch (RemotingSendRequestException ignore) {
         }
@@ -247,7 +253,7 @@ public void noClientAuthFailure() throws Exception {
     @Test
     public void clientRejectsUntrustedServerCert() throws Exception {
         try {
-            RemotingCommand response = remotingClient.invokeSync("localhost:" + remotingServer.localListenPort(), createRequest(), 1000 * 3);
+            RemotingCommand response = remotingClient.invokeSync(getServerAddress(), createRequest(), 1000 * 3);
             failBecauseExceptionWasNotThrown(RemotingSendRequestException.class);
         } catch (RemotingSendRequestException ignore) {
         }
@@ -333,6 +339,10 @@ private static String getCertsPath(String fileName) {
         }
     }
 
+    private String getServerAddress() {
+        return "localhost:" + remotingServer.localListenPort();
+    }
+
     private static RemotingCommand createRequest() {
         RequestHeader requestHeader = new RequestHeader();
         requestHeader.setCount(1);
@@ -345,10 +355,19 @@ private void requestThenAssertResponse() throws Exception {
     }
 
     private void requestThenAssertResponse(RemotingClient remotingClient) throws Exception {
-        RemotingCommand response = remotingClient.invokeSync("localhost:" + remotingServer.localListenPort(), createRequest(), 1000 * 3);
+        RemotingCommand response = remotingClient.invokeSync(getServerAddress(), createRequest(), 1000 * 3);
         assertNotNull(response);
         assertThat(response.getLanguage()).isEqualTo(LanguageCode.JAVA);
         assertThat(response.getExtFields()).hasSize(2);
         assertThat(response.getExtFields().get("messageTitle")).isEqualTo("Welcome");
     }
+
+    private boolean isHostConnectable(String addr) {
+        try (Socket socket = new Socket()) {
+            socket.connect(RemotingUtil.string2SocketAddress(addr));
+            return true;
+        } catch (IOException ignored) {
+        }
+        return false;
+    }
 }
diff --git a/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstractTest.java b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstractTest.java
@@ -105,4 +105,21 @@ public void operationComplete(final ResponseFuture responseFuture) {
         remotingAbstract.scanResponseTable();
         assertNull(remotingAbstract.responseTable.get(dummyId));
     }
+
+    @Test
+    public void testProcessRequestCommand() throws InterruptedException {
+        final Semaphore semaphore = new Semaphore(0);
+        RemotingCommand request = RemotingCommand.createRequestCommand(1, null);
+        ResponseFuture responseFuture = new ResponseFuture(null, 1, request, 3000,
+            responseFuture1 -> assertThat(semaphore.availablePermits()).isEqualTo(0), new SemaphoreReleaseOnlyOnce(semaphore));
+
+        remotingAbstract.responseTable.putIfAbsent(1, responseFuture);
+        RemotingCommand response = RemotingCommand.createResponseCommand(0, "Foo");
+        response.setOpaque(1);
+        remotingAbstract.processResponseCommand(null, response);
+
+        // Acquire the release permit after call back
+        semaphore.acquire(1);
+        assertThat(semaphore.availablePermits()).isEqualTo(0);
+    }
 }