diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index e660a00..43771b5 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -16,7 +16,7 @@ jobs: uses: actions/checkout@v4 - name: Set up JDK 8 - uses: actions/setup-java@v3 + uses: actions/setup-java@v4 with: distribution: 'temurin' java-version: 8 @@ -24,9 +24,9 @@ jobs: - name: Check Maven module version run: | VERSION=$(mvn help:evaluate -Dexpression=project.version -q -DforceStdout) - if [[ $GITHUB_REF != 'refs/heads/main' && $VERSION != *"-SNAPSHOT"* ]]; then - echo "Skipping release version $VERSION on non-main branch" - exit 0 + if [[ $GITHUB_REF == 'refs/heads/develop' && $VERSION != *"-SNAPSHOT"* ]]; then + echo "Refusing to deploy release version $VERSION from develop" + exit 1 fi - name: Import GPG private key diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 0000000..232ef35 --- /dev/null +++ b/.github/workflows/test.yml @@ -0,0 +1,37 @@ +name: Test + +on: + pull_request: + push: + branches: + - main + - develop + +permissions: + contents: read + +jobs: + test: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + java: + - '8' + - '17' + - '21' + - '24' + + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up JDK + uses: actions/setup-java@v4 + with: + distribution: temurin + java-version: ${{ matrix.java }} + cache: maven + + - name: Test and package + run: mvn -B -ntp package diff --git a/app-stream-api/pom.xml b/app-stream-api/pom.xml index ff5f324..152b0ea 100644 --- a/app-stream-api/pom.xml +++ b/app-stream-api/pom.xml @@ -5,7 +5,7 @@ open-app-stream-client com.dingtalk.open - 1.3.13 + 1.3.14 ../pom.xml 4.0.0 diff --git a/app-stream-api/src/main/java/com/dingtalk/open/app/api/OpenDingTalkStreamClient.java b/app-stream-api/src/main/java/com/dingtalk/open/app/api/OpenDingTalkStreamClient.java index 13c2309..3341d7d 100644 --- a/app-stream-api/src/main/java/com/dingtalk/open/app/api/OpenDingTalkStreamClient.java +++ b/app-stream-api/src/main/java/com/dingtalk/open/app/api/OpenDingTalkStreamClient.java @@ -11,13 +11,18 @@ import com.dingtalk.open.app.stream.network.api.ClientConnectionListener; import com.dingtalk.open.app.stream.network.api.EndPointConnection; import com.dingtalk.open.app.stream.network.api.NetProxy; +import com.dingtalk.open.app.stream.network.api.NetworkSharedResources; +import com.dingtalk.open.app.stream.network.api.logger.InternalLogger; +import com.dingtalk.open.app.stream.network.api.logger.InternalLoggerFactory; import com.dingtalk.open.app.stream.network.core.EndPointConnectionFactory; import com.dingtalk.open.app.stream.network.core.NetWorkService; import com.dingtalk.open.app.stream.network.core.Subscription; import java.util.Collections; +import java.util.HashSet; import java.util.Set; import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; @@ -26,6 +31,8 @@ * @date 2022/12/26 */ class OpenDingTalkStreamClient implements OpenDingTalkClient { + private static final InternalLogger LOGGER = InternalLoggerFactory.getLogger(OpenDingTalkStreamClient.class); + private static final long EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS = 5L; private final DingTalkCredential credential; private final CommandDispatcher dispatcher; private final ExecutorService executor; @@ -35,6 +42,7 @@ class OpenDingTalkStreamClient implements OpenDingTalkClient { private NetWorkService netWorkService; private OpenApiClient openApiClient; private Set subscriptions; + private boolean networkResourcesAcquired; public OpenDingTalkStreamClient(DingTalkCredential credential, CommandDispatcher dispatcher, ExecutorService executor, ClientOption option, Set subscriptions, NetProxy netProxy) { @@ -42,7 +50,7 @@ public OpenDingTalkStreamClient(DingTalkCredential credential, CommandDispatcher this.dispatcher = dispatcher; this.executor = executor; this.option = option; - this.subscriptions = Collections.unmodifiableSet(subscriptions); + this.subscriptions = Collections.unmodifiableSet(new HashSet<>(subscriptions)); this.status = new AtomicReference<>(Status.INIT); this.netProxy = netProxy; } @@ -50,12 +58,33 @@ public OpenDingTalkStreamClient(DingTalkCredential credential, CommandDispatcher @Override public synchronized void start() throws OpenDingTalkAppException { if (status.get() == Status.INIT) { - this.openApiClient = OpenApiClientBuilder.create().setHost(option.getOpenApiHost()).setTimeout(option.getConnectionTTL()).setProxy(netProxy).build(); - final EndPointConnectionFactory factory = () -> openConnection(this.credential, subscriptions, netProxy); - ClientConnectionListener listener = new AppServiceListener(dispatcher, executor); - this.netWorkService = new NetWorkService(factory, listener, option.getMaxConnectionCount(), option.getConnectionTTL(), option.getConnectTimeout(), option.getKeepAliveOption().getKeepAliveIdleMill()); - this.netWorkService.start(); - this.status.set(Status.ACTIVE); + NetworkSharedResources.acquireNetWorkEventLoopGroup(); + networkResourcesAcquired = true; + try { + this.openApiClient = OpenApiClientBuilder.create() + .setHost(option.getOpenApiHost()) + .setTimeout(toHttpTimeout(option.getConnectTimeout())) + .setProxy(netProxy) + .build(); + final EndPointConnectionFactory factory = () -> openConnection(this.credential, subscriptions, netProxy); + ClientConnectionListener listener = new AppServiceListener(dispatcher, executor); + this.netWorkService = new NetWorkService(factory, listener, option.getMaxConnectionCount(), option.getConnectionTTL(), option.getConnectTimeout(), option.getKeepAliveOption().getKeepAliveIdleMill()); + this.netWorkService.start(); + this.status.set(Status.ACTIVE); + } catch (RuntimeException | Error e) { + try { + if (this.netWorkService != null) { + this.netWorkService.shutdown(); + } + } catch (Exception shutdownError) { + e.addSuppressed(shutdownError); + } finally { + shutdownConsumerExecutor(); + releaseNetworkResources(); + status.set(Status.INACTIVE); + } + throw e; + } } else if (status.get() == Status.INACTIVE) { throw new OpenDingTalkAppException(DingTalkAppError.CLIENT_STATE_ERROR); } @@ -63,17 +92,49 @@ public synchronized void start() throws OpenDingTalkAppException { @Override public synchronized void stop() throws Exception { - if (status.get() == Status.ACTIVE) { - if (this.netWorkService != null) { - this.netWorkService.shutdown(); + if (status.get() != Status.INACTIVE) { + try { + if (this.netWorkService != null) { + this.netWorkService.shutdown(); + } + } finally { + try { + shutdownConsumerExecutor(); + } finally { + releaseNetworkResources(); + status.set(Status.INACTIVE); + } } - if (executor != null) { - this.executor.shutdown(); + } + } + + private void shutdownConsumerExecutor() { + if (executor == null) { + return; + } + this.executor.shutdownNow(); + try { + if (!this.executor.awaitTermination(EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) { + LOGGER.warn("[DingTalk] consumer executor did not terminate within {} seconds", + EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS); } - status.set(Status.INACTIVE); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.warn("[DingTalk] interrupted while waiting for consumer executor shutdown"); } } + private void releaseNetworkResources() { + if (networkResourcesAcquired) { + NetworkSharedResources.releaseNetWorkEventLoopGroup(); + networkResourcesAcquired = false; + } + } + + static int toHttpTimeout(long timeout) { + return (int) Math.min(timeout, Integer.MAX_VALUE); + } + private EndPointConnection openConnection(DingTalkCredential credential, Set subscriptions, NetProxy proxy) throws Exception { OpenConnectionRequest request = new OpenConnectionRequest(); request.setClientId(credential.getClientId()); diff --git a/app-stream-api/src/main/java/com/dingtalk/open/app/api/OpenDingTalkStreamClientBuilder.java b/app-stream-api/src/main/java/com/dingtalk/open/app/api/OpenDingTalkStreamClientBuilder.java index 46936d6..0075bfe 100644 --- a/app-stream-api/src/main/java/com/dingtalk/open/app/api/OpenDingTalkStreamClientBuilder.java +++ b/app-stream-api/src/main/java/com/dingtalk/open/app/api/OpenDingTalkStreamClientBuilder.java @@ -90,7 +90,7 @@ public OpenDingTalkStreamClientBuilder timeout(int timeout) { public OpenDingTalkStreamClientBuilder connectTimeout(long connectTimeout) { - this.connectTimeout = connectTimeout; + this.connectTimeout = Preconditions.checkPositive(connectTimeout); return this; } @@ -154,6 +154,7 @@ public OpenDingTalkStreamClientBuilder forwardGraphRequestToHTTP(int port) { public OpenDingTalkClient build() { + DingTalkCredential validCredential = Preconditions.notNull(credential); ClientOption option = new ClientOption(); option.setConnectTimeout(connectTimeout); option.setMaxConnectionCount(maxConnectionCount); @@ -161,7 +162,8 @@ public OpenDingTalkClient build() { option.setOpenApiHost(openApiHost); option.setKeepAliveOption(keepAliveOption); ExecutorService executor = ThreadUtil.newFixedExecutor(consumeThreads, "DingTalk-Consumer"); - return new OpenDingTalkStreamClient(credential, new CommandDispatcher(commands), executor, option, subscriptions, netProxy); + return new OpenDingTalkStreamClient(validCredential, new CommandDispatcher(commands), + executor, option, subscriptions, netProxy); } private void subscribe(CommandType type, String topic) { diff --git a/app-stream-api/src/main/java/com/dingtalk/open/app/api/command/CommandDispatcher.java b/app-stream-api/src/main/java/com/dingtalk/open/app/api/command/CommandDispatcher.java index 50e31b0..9af7ada 100644 --- a/app-stream-api/src/main/java/com/dingtalk/open/app/api/command/CommandDispatcher.java +++ b/app-stream-api/src/main/java/com/dingtalk/open/app/api/command/CommandDispatcher.java @@ -9,6 +9,7 @@ import com.dingtalk.open.app.stream.protocol.CommandType; import java.util.Collections; +import java.util.HashMap; import java.util.Map; /** @@ -20,8 +21,9 @@ public class CommandDispatcher { private final Map registry; public CommandDispatcher(Map registry) { - registry.putIfAbsent(CommandType.SYSTEM, new SystemCommandExecutor()); - this.registry = Collections.unmodifiableMap(registry); + Map registrySnapshot = new HashMap<>(registry); + registrySnapshot.putIfAbsent(CommandType.SYSTEM, new SystemCommandExecutor()); + this.registry = Collections.unmodifiableMap(registrySnapshot); } /** diff --git a/app-stream-api/src/main/java/com/dingtalk/open/app/api/open/HttpOpenApiClient.java b/app-stream-api/src/main/java/com/dingtalk/open/app/api/open/HttpOpenApiClient.java index 4813631..89d7c20 100644 --- a/app-stream-api/src/main/java/com/dingtalk/open/app/api/open/HttpOpenApiClient.java +++ b/app-stream-api/src/main/java/com/dingtalk/open/app/api/open/HttpOpenApiClient.java @@ -7,10 +7,20 @@ import com.dingtalk.open.app.api.util.IoUtils; import com.dingtalk.open.app.stream.network.api.NetProxy; +import java.io.InputStream; +import java.io.OutputStream; +import java.lang.reflect.Field; +import java.lang.reflect.Method; import java.net.Authenticator; import java.net.HttpURLConnection; +import java.net.InetAddress; +import java.net.InetSocketAddress; import java.net.PasswordAuthentication; +import java.net.Proxy; import java.net.URL; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.concurrent.Callable; /** * @author feiyin @@ -18,6 +28,31 @@ */ class HttpOpenApiClient implements OpenApiClient { + /* + * Java 8 HttpURLConnection only supports authenticated HTTPS proxies via + * the JVM-wide Authenticator. Keep the unavoidable global hook inert + * outside one SDK request, match the exact proxy endpoint, and serialize + * authenticated proxy requests so credentials from different clients can + * never overwrite one another. + */ + private static final Object PROXY_AUTHENTICATION_LOCK = new Object(); + private static volatile ProxyAuthenticationContext activeProxyAuthentication; + private static final Authenticator PROXY_AUTHENTICATOR = new Authenticator() { + @Override + protected PasswordAuthentication getPasswordAuthentication() { + ProxyAuthenticationContext authentication = activeProxyAuthentication; + if (getRequestorType() != RequestorType.PROXY + || authentication == null + || !authentication.matches( + getRequestingHost(), + getRequestingSite(), + getRequestingPort())) { + return null; + } + return authentication.passwordAuthentication(); + } + }; + private final String host; private final NetProxy netProxy; @@ -33,37 +68,150 @@ public HttpOpenApiClient(String host, int timeout, NetProxy netProxy) { @Override public OpenConnectionResponse openConnection(OpenConnectionRequest request) throws Exception { URL url = new URL(host + "/v1.0/gateway/connections/open"); + if (hasProxyCredentials(netProxy)) { + return withProxyAuthentication( + netProxy, + () -> executeOpenConnection(url, request)); + } + return executeOpenConnection(url, request); + } + private OpenConnectionResponse executeOpenConnection( + URL url, + OpenConnectionRequest request) throws Exception { HttpURLConnection connection; if (netProxy != null) { - if (netProxy.getUsername() != null && netProxy.getPassword() != null) { - Authenticator.setDefault(new Authenticator() { - @Override - protected PasswordAuthentication getPasswordAuthentication() { - return new PasswordAuthentication(netProxy.getUsername(), netProxy.getPassword().toCharArray()); - } - }); - } connection = (HttpURLConnection) url.openConnection(netProxy.getProxy()); } else { connection = (HttpURLConnection) url.openConnection(); } - connection.setRequestMethod(HttpConstants.METHOD_POST); - connection.setReadTimeout(this.timeout); - connection.setConnectTimeout(this.timeout); - connection.setRequestProperty(HttpConstants.HEADER_CONTENT_TYPE, HttpConstants.CONTENT_TYPE_APPLICATION_JSON); - connection.setRequestProperty(HttpConstants.HEADER_ACCEPT, "application/json"); - connection.setDoInput(true); - connection.setDoOutput(true); - connection.connect(); - connection.getOutputStream().write(JSON.toJSONBytes(request, JSONWriter.Feature.WriteEnumUsingToString)); - connection.getOutputStream().flush(); - if (connection.getResponseCode() == HttpConstants.STATUS_OK) { - byte[] content = IoUtils.readAll(connection.getInputStream()); - return JSON.parseObject(content, OpenConnectionResponse.class); - } else { - byte[] content = IoUtils.readAll(connection.getErrorStream()); - throw DingTalkAppError.HTTP_ERROR_RESPONSE.toException(String.format("status=%s,msg=%s", connection.getResponseCode(), content != null ? new String(content) : "")); + try { + connection.setRequestMethod(HttpConstants.METHOD_POST); + connection.setReadTimeout(this.timeout); + connection.setConnectTimeout(this.timeout); + connection.setRequestProperty(HttpConstants.HEADER_CONTENT_TYPE, HttpConstants.CONTENT_TYPE_APPLICATION_JSON); + connection.setRequestProperty(HttpConstants.HEADER_ACCEPT, "application/json"); + connection.setDoInput(true); + connection.setDoOutput(true); + connection.connect(); + try (OutputStream output = connection.getOutputStream()) { + output.write(JSON.toJSONBytes(request, JSONWriter.Feature.WriteEnumUsingToString)); + output.flush(); + } + + int status = connection.getResponseCode(); + InputStream responseStream = status == HttpConstants.STATUS_OK + ? connection.getInputStream() + : connection.getErrorStream(); + byte[] content; + if (responseStream == null) { + content = null; + } else { + try (InputStream input = responseStream) { + content = IoUtils.readAll(input); + } + } + + if (status == HttpConstants.STATUS_OK) { + return JSON.parseObject(content, OpenConnectionResponse.class); + } + throw DingTalkAppError.HTTP_ERROR_RESPONSE.toException(String.format( + "status=%s,msg=%s", + status, + content != null ? new String(content, StandardCharsets.UTF_8) : "")); + } finally { + connection.disconnect(); + } + } + + static T withProxyAuthentication( + NetProxy netProxy, + Callable operation) throws Exception { + if (!hasProxyCredentials(netProxy)) { + return operation.call(); + } + synchronized (PROXY_AUTHENTICATION_LOCK) { + ProxyAuthenticationContext authentication = + new ProxyAuthenticationContext(netProxy); + Authenticator previousAuthenticator = getDefaultAuthenticator(); + activeProxyAuthentication = authentication; + Authenticator.setDefault(PROXY_AUTHENTICATOR); + try { + return operation.call(); + } finally { + activeProxyAuthentication = null; + authentication.clear(); + Authenticator.setDefault(previousAuthenticator); + } + } + } + + private static Authenticator getDefaultAuthenticator() { + try { + // Authenticator.getDefault() is public starting with Java 9. + Method getDefault = Authenticator.class.getMethod("getDefault"); + return (Authenticator) getDefault.invoke(null); + } catch (NoSuchMethodException ignored) { + try { + // Java 8 has no public accessor. Read the same private field + // used by setDefault so the host application's authenticator + // can still be restored after this short request scope. + Field field = Authenticator.class.getDeclaredField("theAuthenticator"); + field.setAccessible(true); + return (Authenticator) field.get(null); + } catch (ReflectiveOperationException | SecurityException e) { + throw new IllegalStateException( + "cannot preserve the JVM default Authenticator", e); + } + } catch (ReflectiveOperationException | SecurityException e) { + throw new IllegalStateException( + "cannot read the JVM default Authenticator", e); + } + } + + private static boolean hasProxyCredentials(NetProxy netProxy) { + return netProxy != null + && netProxy.getUsername() != null + && netProxy.getPassword() != null; + } + + private static class ProxyAuthenticationContext { + private final String host; + private final InetAddress address; + private final int port; + private final String username; + private final char[] password; + + private ProxyAuthenticationContext(NetProxy netProxy) { + Proxy proxy = netProxy.getProxy(); + InetSocketAddress socketAddress = (InetSocketAddress) proxy.address(); + this.host = socketAddress.getHostString(); + this.address = socketAddress.getAddress(); + this.port = socketAddress.getPort(); + this.username = netProxy.getUsername(); + this.password = netProxy.getPassword().toCharArray(); + } + + private boolean matches( + String requestingHost, + InetAddress requestingAddress, + int requestingPort) { + if (requestingPort != port) { + return false; + } + if (requestingHost != null && host.equalsIgnoreCase(requestingHost)) { + return true; + } + return address != null && address.equals(requestingAddress); + } + + private PasswordAuthentication passwordAuthentication() { + return new PasswordAuthentication(username, password); + } + + private void clear() { + Arrays.fill(password, '\0'); } } + } diff --git a/app-stream-api/src/main/java/com/dingtalk/open/app/api/protocol/AppServiceListener.java b/app-stream-api/src/main/java/com/dingtalk/open/app/api/protocol/AppServiceListener.java index 66cb718..9c30d34 100644 --- a/app-stream-api/src/main/java/com/dingtalk/open/app/api/protocol/AppServiceListener.java +++ b/app-stream-api/src/main/java/com/dingtalk/open/app/api/protocol/AppServiceListener.java @@ -6,8 +6,12 @@ import com.dingtalk.open.app.stream.network.api.Context; import com.dingtalk.open.app.stream.network.api.logger.InternalLogger; import com.dingtalk.open.app.stream.network.api.logger.InternalLoggerFactory; +import com.dingtalk.open.app.stream.protocol.CommandType; +import com.dingtalk.open.app.stream.protocol.event.AckPayload; +import com.dingtalk.open.app.stream.protocol.event.EventAckStatus; import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; /** * @author feiyin @@ -25,12 +29,33 @@ public AppServiceListener(CommandDispatcher commandDispatcher, Executor executor @Override public void receive(Context context) { - this.executor.execute(() -> { - try { - commandDispatcher.execute(context); - } catch (Exception e) { - LOGGER.error("[DingTalk] dispatch command failed, {}", e); + if (context.getRequest().getType() == CommandType.SYSTEM) { + dispatch(context); + return; + } + try { + this.executor.execute(() -> dispatch(context)); + } catch (RejectedExecutionException e) { + if (context.getRequest().getType() == CommandType.EVENT) { + AckPayload ackPayload = new AckPayload(); + ackPayload.setStatus(EventAckStatus.LATER); + ackPayload.setMessage("consumer capacity reached"); + context.replay(ackPayload); + } else { + // CALLBACK responses are application-specific. Leaving the + // frame unacknowledged lets the server retry it later. + LOGGER.warn("[DingTalk] callback consumer capacity reached, topic={}", + context.getRequest().getTopic()); } - }); + } + } + + private void dispatch(Context context) { + try { + commandDispatcher.execute(context); + } catch (Throwable e) { + LOGGER.error("[DingTalk] dispatch command failed, {}", e); + context.exception(e); + } } } diff --git a/app-stream-api/src/test/java/com/dingtalk/open/app/api/OpenDingTalkStreamClientTest.java b/app-stream-api/src/test/java/com/dingtalk/open/app/api/OpenDingTalkStreamClientTest.java new file mode 100644 index 0000000..587dfab --- /dev/null +++ b/app-stream-api/src/test/java/com/dingtalk/open/app/api/OpenDingTalkStreamClientTest.java @@ -0,0 +1,74 @@ +package com.dingtalk.open.app.api; + +import com.dingtalk.open.app.api.command.CommandDispatcher; +import com.dingtalk.open.app.api.security.DingTalkCredential; +import org.junit.Assert; +import org.junit.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +public class OpenDingTalkStreamClientTest { + + @Test + public void openApiTimeoutUsesConnectTimeoutRange() { + Assert.assertEquals(3_000, OpenDingTalkStreamClient.toHttpTimeout(3_000L)); + Assert.assertEquals( + Integer.MAX_VALUE, + OpenDingTalkStreamClient.toHttpTimeout((long) Integer.MAX_VALUE + 1L)); + } + + @Test(expected = OpenDingTalkAppException.class) + public void connectTimeoutMustBePositive() { + OpenDingTalkStreamClientBuilder.custom().connectTimeout(0); + } + + @Test(expected = OpenDingTalkAppException.class) + public void buildRequiresCredentialBeforeCreatingExecutor() { + OpenDingTalkStreamClientBuilder.custom().build(); + } + + @Test + public void stopBeforeStartTerminatesConsumerExecutor() throws Exception { + ExecutorService executor = Executors.newSingleThreadExecutor(); + CountDownLatch taskStarted = new CountDownLatch(1); + executor.execute(() -> { + taskStarted.countDown(); + try { + new CountDownLatch(1).await(); + } catch (InterruptedException ignored) { + Thread.currentThread().interrupt(); + } + }); + Assert.assertTrue(taskStarted.await(1, TimeUnit.SECONDS)); + + OpenDingTalkStreamClient client = new OpenDingTalkStreamClient( + new TestCredential(), + new CommandDispatcher(new HashMap<>()), + executor, + new ClientOption(), + Collections.emptySet(), + null); + client.stop(); + + Assert.assertTrue(executor.isShutdown()); + Assert.assertTrue(executor.awaitTermination(1, TimeUnit.SECONDS)); + Assert.assertTrue(executor.isTerminated()); + } + + private static class TestCredential implements DingTalkCredential { + @Override + public String getClientId() { + return "client-id"; + } + + @Override + public String getClientSecret() { + return "client-secret"; + } + } +} diff --git a/app-stream-api/src/test/java/com/dingtalk/open/app/api/open/HttpOpenApiClientTest.java b/app-stream-api/src/test/java/com/dingtalk/open/app/api/open/HttpOpenApiClientTest.java new file mode 100644 index 0000000..e6d3ad6 --- /dev/null +++ b/app-stream-api/src/test/java/com/dingtalk/open/app/api/open/HttpOpenApiClientTest.java @@ -0,0 +1,179 @@ +package com.dingtalk.open.app.api.open; + +import com.dingtalk.open.app.stream.network.api.NetProxy; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import java.net.Authenticator; +import java.net.InetAddress; +import java.net.PasswordAuthentication; +import java.net.URL; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +public class HttpOpenApiClientTest { + + private static final String PROXY_HOST = "127.0.0.1"; + private static final int PROXY_PORT = 3128; + + @Before + public void clearDefaultAuthenticatorBeforeTest() { + Authenticator.setDefault(null); + } + + @After + public void clearDefaultAuthenticatorAfterTest() { + Authenticator.setDefault(null); + } + + @Test + public void credentialsAreScopedToMatchingProxyRequests() throws Exception { + NetProxy proxy = new NetProxy(PROXY_HOST, PROXY_PORT, "proxy-user", "proxy-password"); + + HttpOpenApiClient.withProxyAuthentication(proxy, () -> { + PasswordAuthentication matching = requestAuthentication( + PROXY_HOST, + InetAddress.getByName(PROXY_HOST), + PROXY_PORT, + Authenticator.RequestorType.PROXY); + Assert.assertNotNull(matching); + Assert.assertEquals("proxy-user", matching.getUserName()); + Assert.assertArrayEquals("proxy-password".toCharArray(), matching.getPassword()); + + Assert.assertNull(requestAuthentication( + PROXY_HOST, + InetAddress.getByName(PROXY_HOST), + PROXY_PORT, + Authenticator.RequestorType.SERVER)); + Assert.assertNull(requestAuthentication( + "127.0.0.2", + InetAddress.getByName("127.0.0.2"), + PROXY_PORT, + Authenticator.RequestorType.PROXY)); + Assert.assertNull(requestAuthentication( + PROXY_HOST, + InetAddress.getByName(PROXY_HOST), + PROXY_PORT + 1, + Authenticator.RequestorType.PROXY)); + return null; + }); + + Assert.assertNull(requestAuthentication( + PROXY_HOST, + InetAddress.getByName(PROXY_HOST), + PROXY_PORT, + Authenticator.RequestorType.PROXY)); + } + + @Test + public void credentialsAreClearedWhenRequestFails() throws Exception { + NetProxy proxy = new NetProxy(PROXY_HOST, PROXY_PORT, "proxy-user", "proxy-password"); + Exception expected = new Exception("simulated request failure"); + + try { + HttpOpenApiClient.withProxyAuthentication(proxy, () -> { + throw expected; + }); + Assert.fail("request failure should propagate"); + } catch (Exception actual) { + Assert.assertSame(expected, actual); + } + + Assert.assertNull(requestAuthentication( + PROXY_HOST, + InetAddress.getByName(PROXY_HOST), + PROXY_PORT, + Authenticator.RequestorType.PROXY)); + } + + @Test + public void hostAuthenticatorIsRestoredAfterProxyRequest() throws Exception { + Authenticator hostAuthenticator = new Authenticator() { + @Override + protected PasswordAuthentication getPasswordAuthentication() { + return new PasswordAuthentication( + "host-user", + "host-password".toCharArray()); + } + }; + Authenticator.setDefault(hostAuthenticator); + NetProxy proxy = new NetProxy(PROXY_HOST, PROXY_PORT, "proxy-user", "proxy-password"); + + HttpOpenApiClient.withProxyAuthentication(proxy, () -> { + PasswordAuthentication proxyAuthentication = requestAuthentication( + PROXY_HOST, + InetAddress.getByName(PROXY_HOST), + PROXY_PORT, + Authenticator.RequestorType.PROXY); + Assert.assertNotNull(proxyAuthentication); + Assert.assertEquals("proxy-user", proxyAuthentication.getUserName()); + return null; + }); + + PasswordAuthentication restored = requestAuthentication( + "service.example", + InetAddress.getByName(PROXY_HOST), + 443, + Authenticator.RequestorType.SERVER); + Assert.assertNotNull(restored); + Assert.assertEquals("host-user", restored.getUserName()); + Assert.assertArrayEquals( + "host-password".toCharArray(), + restored.getPassword()); + } + + @Test + public void concurrentClientsCannotObserveEachOthersProxyCredentials() throws Exception { + ExecutorService executor = Executors.newFixedThreadPool(8); + List> futures = new ArrayList<>(); + try { + for (int index = 0; index < 100; index++) { + final String username = "proxy-user-" + index; + final String password = "proxy-password-" + index; + futures.add(executor.submit(() -> HttpOpenApiClient.withProxyAuthentication( + new NetProxy(PROXY_HOST, PROXY_PORT, username, password), + () -> { + PasswordAuthentication authentication = requestAuthentication( + PROXY_HOST, + InetAddress.getByName(PROXY_HOST), + PROXY_PORT, + Authenticator.RequestorType.PROXY); + Assert.assertNotNull(authentication); + Assert.assertEquals(username, authentication.getUserName()); + Assert.assertArrayEquals( + password.toCharArray(), + authentication.getPassword()); + return null; + }))); + } + for (Future future : futures) { + future.get(5, TimeUnit.SECONDS); + } + } finally { + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + private static PasswordAuthentication requestAuthentication( + String host, + InetAddress address, + int port, + Authenticator.RequestorType requestorType) throws Exception { + return Authenticator.requestPasswordAuthentication( + host, + address, + port, + "https", + "proxy authentication", + "basic", + new URL("https://api.dingtalk.com/v1.0/gateway/connections/open"), + requestorType); + } +} diff --git a/app-stream-api/src/test/java/com/dingtalk/open/app/api/protocol/AppServiceListenerTest.java b/app-stream-api/src/test/java/com/dingtalk/open/app/api/protocol/AppServiceListenerTest.java new file mode 100644 index 0000000..88aeea0 --- /dev/null +++ b/app-stream-api/src/test/java/com/dingtalk/open/app/api/protocol/AppServiceListenerTest.java @@ -0,0 +1,126 @@ +package com.dingtalk.open.app.api.protocol; + +import com.dingtalk.open.app.api.command.CommandDispatcher; +import com.dingtalk.open.app.stream.network.api.Context; +import com.dingtalk.open.app.stream.protocol.CommandType; +import com.dingtalk.open.app.stream.protocol.ProtocolRequest; +import com.dingtalk.open.app.stream.protocol.ProtocolRequestFacade; +import com.dingtalk.open.app.stream.protocol.event.AckPayload; +import com.dingtalk.open.app.stream.protocol.event.EventAckStatus; +import org.junit.Assert; +import org.junit.Test; + +import java.util.HashMap; +import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; + +public class AppServiceListenerTest { + + private static final Executor REJECTING_EXECUTOR = command -> { + throw new RejectedExecutionException("full"); + }; + + @Test + public void overloadedEventIsAcknowledgedForLaterRetry() { + TestContext context = new TestContext(CommandType.EVENT, "event-topic", "{}"); + AppServiceListener listener = new AppServiceListener( + new CommandDispatcher(new HashMap<>()), + REJECTING_EXECUTOR); + + listener.receive(context); + + Assert.assertTrue(context.response instanceof AckPayload); + Assert.assertEquals(EventAckStatus.LATER, ((AckPayload) context.response).getStatus()); + Assert.assertNull(context.failure); + } + + @Test + public void systemCommandBypassesSaturatedConsumerExecutor() { + TestContext context = new TestContext(CommandType.SYSTEM, "ping", "{}"); + AppServiceListener listener = new AppServiceListener( + new CommandDispatcher(new HashMap<>()), + REJECTING_EXECUTOR); + + listener.receive(context); + + Assert.assertNotNull(context.response); + Assert.assertNull(context.failure); + } + + private static class TestContext implements Context { + private final ProtocolRequestFacade request; + private Object response; + private Throwable failure; + + private TestContext(CommandType type, String topic, String data) { + this.request = new TestRequest(type, topic, data); + } + + @Override + public void replay(Object payload) { + response = payload; + } + + @Override + public void exception(Throwable t) { + failure = t; + } + + @Override + public String connectionId() { + return "test-connection"; + } + + @Override + public ProtocolRequestFacade getRequest() { + return request; + } + } + + private static class TestRequest implements ProtocolRequestFacade { + private final CommandType type; + private final String topic; + private final String data; + + private TestRequest(CommandType type, String topic, String data) { + this.type = type; + this.topic = topic; + this.data = data; + } + + @Override + public String getMessageId() { + return "test-message"; + } + + @Override + public String getContentType() { + return "application/json"; + } + + @Override + public String getTopic() { + return topic; + } + + @Override + public CommandType getType() { + return type; + } + + @Override + public String getData() { + return data; + } + + @Override + public String getHeader(String headerName) { + return null; + } + + @Override + public ProtocolRequest getRequest() { + return null; + } + } +} diff --git a/app-stream-client/pom.xml b/app-stream-client/pom.xml index fcdb837..c7f943c 100644 --- a/app-stream-client/pom.xml +++ b/app-stream-client/pom.xml @@ -4,13 +4,13 @@ com.dingtalk.open open-app-stream-client - 1.3.13 + 1.3.14 ../pom.xml app-stream-client jar - 1.3.13 + 1.3.14 app-stream-client @@ -101,11 +101,7 @@ - - META-INF/services/com.dingtalk.open.app.stream.network.rsocket.RsocketTransportConnector - - + implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/> diff --git a/app-stream-network/app-stream-network-api/pom.xml b/app-stream-network/app-stream-network-api/pom.xml index faa0c84..3225248 100644 --- a/app-stream-network/app-stream-network-api/pom.xml +++ b/app-stream-network/app-stream-network-api/pom.xml @@ -4,17 +4,22 @@ com.dingtalk.open app-stream-network - 1.3.13 + 1.3.14 ../pom.xml app-stream-network-api - 1.3.13 + 1.3.14 jar app-stream-network-api + + junit + junit + test + com.dingtalk.open app-stream-protocol diff --git a/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/NetworkSharedResources.java b/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/NetworkSharedResources.java index 219dbed..26a1971 100644 --- a/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/NetworkSharedResources.java +++ b/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/NetworkSharedResources.java @@ -11,14 +11,31 @@ public class NetworkSharedResources { private static NioEventLoopGroup EVENT_LOOP_GROUP; + private static int referenceCount; - static { - EVENT_LOOP_GROUP = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2); + public static synchronized EventLoopGroup acquireNetWorkEventLoopGroup() { + referenceCount++; + return getNetWorkEventLoopGroup(); } - public static EventLoopGroup getNetWorkEventLoopGroup() { + public static synchronized EventLoopGroup getNetWorkEventLoopGroup() { + if (EVENT_LOOP_GROUP == null + || EVENT_LOOP_GROUP.isShuttingDown() + || EVENT_LOOP_GROUP.isShutdown() + || EVENT_LOOP_GROUP.isTerminated()) { + EVENT_LOOP_GROUP = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2); + } return EVENT_LOOP_GROUP; } - + public static synchronized void releaseNetWorkEventLoopGroup() { + if (referenceCount == 0) { + return; + } + referenceCount--; + if (referenceCount == 0 && EVENT_LOOP_GROUP != null) { + EVENT_LOOP_GROUP.shutdownGracefully(); + EVENT_LOOP_GROUP = null; + } + } } diff --git a/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/exception/DingTalkNetworkException.java b/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/exception/DingTalkNetworkException.java index deb3929..98c8d86 100644 --- a/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/exception/DingTalkNetworkException.java +++ b/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/exception/DingTalkNetworkException.java @@ -6,9 +6,25 @@ */ public class DingTalkNetworkException extends RuntimeException { + private final NetWorkError netWorkError; public DingTalkNetworkException(NetWorkError netWorkError) { + super(netWorkError != null ? netWorkError.name() : null); + this.netWorkError = netWorkError; + } + + public DingTalkNetworkException(NetWorkError netWorkError, Throwable cause) { + super(netWorkError != null ? netWorkError.name() : null, cause); + this.netWorkError = netWorkError; + } + + public DingTalkNetworkException(NetWorkError netWorkError, String detail) { + super(netWorkError != null ? netWorkError.name() + ": " + detail : detail); + this.netWorkError = netWorkError; + } + public NetWorkError getNetWorkError() { + return netWorkError; } } diff --git a/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/logger/NettyInternalLogger.java b/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/logger/NettyInternalLogger.java index 5eb8ce2..56a17d1 100644 --- a/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/logger/NettyInternalLogger.java +++ b/app-stream-network/app-stream-network-api/src/main/java/com/dingtalk/open/app/stream/network/api/logger/NettyInternalLogger.java @@ -18,7 +18,7 @@ public void info(String format, Object... args) { @Override public void warn(String format, Object... args) { - this.delegate.error(format, args); + this.delegate.warn(format, args); } @Override diff --git a/app-stream-network/app-stream-network-api/src/test/java/com/dingtalk/open/app/stream/network/api/NetworkSharedResourcesTest.java b/app-stream-network/app-stream-network-api/src/test/java/com/dingtalk/open/app/stream/network/api/NetworkSharedResourcesTest.java new file mode 100644 index 0000000..1a9b31c --- /dev/null +++ b/app-stream-network/app-stream-network-api/src/test/java/com/dingtalk/open/app/stream/network/api/NetworkSharedResourcesTest.java @@ -0,0 +1,38 @@ +package com.dingtalk.open.app.stream.network.api; + +import io.netty.channel.EventLoopGroup; +import org.junit.Assert; +import org.junit.Test; + +import java.util.concurrent.TimeUnit; + +public class NetworkSharedResourcesTest { + + @Test + public void sharedEventLoopStopsAfterLastRelease() throws Exception { + EventLoopGroup first = NetworkSharedResources.acquireNetWorkEventLoopGroup(); + EventLoopGroup second = NetworkSharedResources.acquireNetWorkEventLoopGroup(); + Assert.assertSame(first, second); + first.submit(() -> { + // Start a real event-loop thread; an unstarted group can appear to + // shut down correctly without proving that process-exit threads end. + }).syncUninterruptibly(); + + NetworkSharedResources.releaseNetWorkEventLoopGroup(); + Assert.assertFalse("one active client still owns the event loop", first.isShuttingDown()); + + NetworkSharedResources.releaseNetWorkEventLoopGroup(); + Assert.assertTrue("last client release must stop the event loop", first.isShuttingDown()); + Assert.assertTrue( + "started event-loop threads must terminate after the last client releases them", + first.awaitTermination(10, TimeUnit.SECONDS)); + Assert.assertTrue(first.isTerminated()); + + EventLoopGroup replacement = NetworkSharedResources.acquireNetWorkEventLoopGroup(); + try { + Assert.assertNotSame("a stopped event loop must not be reused", first, replacement); + } finally { + NetworkSharedResources.releaseNetWorkEventLoopGroup(); + } + } +} diff --git a/app-stream-network/app-stream-network-api/src/test/java/com/dingtalk/open/app/stream/network/api/exception/DingTalkNetworkExceptionTest.java b/app-stream-network/app-stream-network-api/src/test/java/com/dingtalk/open/app/stream/network/api/exception/DingTalkNetworkExceptionTest.java new file mode 100644 index 0000000..438f3c3 --- /dev/null +++ b/app-stream-network/app-stream-network-api/src/test/java/com/dingtalk/open/app/stream/network/api/exception/DingTalkNetworkExceptionTest.java @@ -0,0 +1,27 @@ +package com.dingtalk.open.app.stream.network.api.exception; + +import org.junit.Assert; +import org.junit.Test; + +public class DingTalkNetworkExceptionTest { + + @Test + public void preservesNetworkErrorAndCause() { + IllegalStateException cause = new IllegalStateException("service loader failed"); + DingTalkNetworkException exception = + new DingTalkNetworkException(NetWorkError.OPEN_CONNECTION_ERROR, cause); + + Assert.assertEquals(NetWorkError.OPEN_CONNECTION_ERROR, exception.getNetWorkError()); + Assert.assertEquals("OPEN_CONNECTION_ERROR", exception.getMessage()); + Assert.assertSame(cause, exception.getCause()); + } + + @Test + public void includesProtocolDetail() { + DingTalkNetworkException exception = + new DingTalkNetworkException(NetWorkError.PROTOCOL_ILLEGAL, "protocol WSS"); + + Assert.assertEquals(NetWorkError.PROTOCOL_ILLEGAL, exception.getNetWorkError()); + Assert.assertEquals("PROTOCOL_ILLEGAL: protocol WSS", exception.getMessage()); + } +} diff --git a/app-stream-network/app-stream-network-api/src/test/java/com/dingtalk/open/app/stream/network/api/logger/NettyInternalLoggerTest.java b/app-stream-network/app-stream-network-api/src/test/java/com/dingtalk/open/app/stream/network/api/logger/NettyInternalLoggerTest.java new file mode 100644 index 0000000..80c3531 --- /dev/null +++ b/app-stream-network/app-stream-network-api/src/test/java/com/dingtalk/open/app/stream/network/api/logger/NettyInternalLoggerTest.java @@ -0,0 +1,28 @@ +package com.dingtalk.open.app.stream.network.api.logger; + +import org.junit.Assert; +import org.junit.Test; + +import java.lang.reflect.Proxy; +import java.util.concurrent.atomic.AtomicReference; + +public class NettyInternalLoggerTest { + + @Test + public void warnDelegatesToWarnLevel() { + AtomicReference invokedMethod = new AtomicReference<>(); + io.netty.util.internal.logging.InternalLogger delegate = + (io.netty.util.internal.logging.InternalLogger) Proxy.newProxyInstance( + getClass().getClassLoader(), + new Class[]{io.netty.util.internal.logging.InternalLogger.class}, + (proxy, method, args) -> { + invokedMethod.set(method.getName()); + return null; + }); + + NettyInternalLogger logger = new NettyInternalLogger(delegate); + logger.warn("recoverable network error, connectionId={}", "conn-test"); + + Assert.assertEquals("warn", invokedMethod.get()); + } +} diff --git a/app-stream-network/app-stream-network-core/pom.xml b/app-stream-network/app-stream-network-core/pom.xml index 98c0785..ec0a627 100644 --- a/app-stream-network/app-stream-network-core/pom.xml +++ b/app-stream-network/app-stream-network-core/pom.xml @@ -4,7 +4,7 @@ com.dingtalk.open app-stream-network - 1.3.13 + 1.3.14 ../pom.xml @@ -26,7 +26,7 @@ com.dingtalk.open app-stream-network-api - ${parent.version} + ${project.parent.version} diff --git a/app-stream-network/app-stream-network-core/src/main/java/com/dingtalk/open/app/stream/network/core/Connector.java b/app-stream-network/app-stream-network-core/src/main/java/com/dingtalk/open/app/stream/network/core/Connector.java index 847ab96..6eb3a76 100644 --- a/app-stream-network/app-stream-network-core/src/main/java/com/dingtalk/open/app/stream/network/core/Connector.java +++ b/app-stream-network/app-stream-network-core/src/main/java/com/dingtalk/open/app/stream/network/core/Connector.java @@ -9,6 +9,7 @@ import java.time.Duration; import java.util.Iterator; import java.util.Map; +import java.util.ServiceConfigurationError; import java.util.ServiceLoader; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.locks.Lock; @@ -39,7 +40,9 @@ public static Session connect(EndPointConnection connection, ClientConnectionLis ensureActive(); TransportConnector connector = CONNECTOR_REGISTRY.get(connection.getProtocol()); if (connector == null) { - throw new DingTalkNetworkException(NetWorkError.PROTOCOL_ILLEGAL); + throw new DingTalkNetworkException( + NetWorkError.PROTOCOL_ILLEGAL, + "no transport connector registered for protocol " + connection.getProtocol()); } ConnectOption option = ConnectOption.builder().setTimeout(timeout).setTtl(ttl).setKeepAliveIdle(Duration.ofMillis(keepAliveIdle)).setKeepAliveTimeout(KEEP_ALIVE_TIMEOUT).build(); return connector.connect(connection, listener, option); @@ -55,7 +58,8 @@ private static void ensureActive() { if (INIT) { return; } - ServiceLoader transportConnectors = ServiceLoader.load(TransportConnector.class); + ServiceLoader transportConnectors = ServiceLoader.load( + TransportConnector.class, Connector.class.getClassLoader()); Iterator it = transportConnectors.iterator(); while (it.hasNext()) { TransportConnector transportConnector = it.next(); @@ -67,9 +71,17 @@ private static void ensureActive() { } } } + if (CONNECTOR_REGISTRY.isEmpty()) { + throw new DingTalkNetworkException( + NetWorkError.OPEN_CONNECTION_ERROR, + "no TransportConnector implementation was found by ServiceLoader"); + } INIT = true; - } catch (Exception e) { + } catch (DingTalkNetworkException e) { + throw e; + } catch (ServiceConfigurationError | RuntimeException e) { LOGGER.error("[DingTalk] client init transport failed, {}", e); + throw new DingTalkNetworkException(NetWorkError.OPEN_CONNECTION_ERROR, e); } finally { INIT_LOCK.unlock(); } diff --git a/app-stream-network/app-stream-network-core/src/main/java/com/dingtalk/open/app/stream/network/core/DefaultSessionPool.java b/app-stream-network/app-stream-network-core/src/main/java/com/dingtalk/open/app/stream/network/core/DefaultSessionPool.java index 0bf97ad..adb6959 100644 --- a/app-stream-network/app-stream-network-core/src/main/java/com/dingtalk/open/app/stream/network/core/DefaultSessionPool.java +++ b/app-stream-network/app-stream-network-core/src/main/java/com/dingtalk/open/app/stream/network/core/DefaultSessionPool.java @@ -24,6 +24,7 @@ public class DefaultSessionPool implements SessionPool { private static final InternalLogger LOGGER = InternalLoggerFactory.getLogger(DefaultSessionPool.class); private static final int MAX_RETRY_COUNT = 3; private static final int INTERVAL = 5 * 1000; + private static final int SHUTDOWN_TIMEOUT_SECONDS = 5; private final ScheduledExecutorService scheduledExecutorService; private final Map sessions; private final AtomicBoolean status; @@ -84,7 +85,7 @@ private void closeSession(String connectionId) { */ private void evict() { for (Map.Entry entry : sessions.entrySet()) { - Session session = sessions.get(entry.getKey()); + Session session = entry.getValue(); if (session.isExpired() || !session.isActive()) { closeSession(session.getId()); } @@ -97,16 +98,23 @@ private void evict() { @Override public void shutdown() { if (status.compareAndSet(true, false)) { - scheduledExecutorService.execute(() -> { - for (String sessionId : sessions.keySet()) { - try { - closeSession(sessionId); - } catch (Exception e) { - LOGGER.error("[DingTalk] close session failed, connectionId={}", sessionId, e); - } + scheduledExecutorService.shutdownNow(); + for (String sessionId : sessions.keySet()) { + try { + closeSession(sessionId); + } catch (Exception e) { + LOGGER.error("[DingTalk] close session failed, connectionId={}", sessionId, e); + } + } + try { + if (!scheduledExecutorService.awaitTermination(SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) { + LOGGER.warn("[DingTalk] connection scheduler did not terminate within {} seconds", + SHUTDOWN_TIMEOUT_SECONDS); } - }); - scheduledExecutorService.execute(scheduledExecutorService::shutdown); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.warn("[DingTalk] interrupted while waiting for connection scheduler shutdown"); + } } } @@ -141,7 +149,16 @@ private void onRequest(String connectionId, ProtocolRequestFacade request) { } session.goAway(); //发起建连 - scheduledExecutorService.execute(new ConnectionTask()); + if (isActive()) { + try { + scheduledExecutorService.execute(new ConnectionTask()); + } catch (RejectedExecutionException e) { + if (isActive()) { + LOGGER.warn("[DingTalk] immediate reconnect task was rejected, connectionId={}", + connectionId); + } + } + } } } } @@ -169,6 +186,10 @@ public void run() { if (session == null) { return; } + if (!isActive()) { + session.close(); + return; + } LOGGER.info("[DingTalk] connection is established, connectionId={}", session.getId()); previous = sessions.put(connection.getConnectionId(), session); if (previous != null) { diff --git a/app-stream-network/app-stream-network-rsocket/pom.xml b/app-stream-network/app-stream-network-rsocket/pom.xml index d455f1a..e955a74 100644 --- a/app-stream-network/app-stream-network-rsocket/pom.xml +++ b/app-stream-network/app-stream-network-rsocket/pom.xml @@ -4,7 +4,7 @@ com.dingtalk.open app-stream-network - 1.3.13 + 1.3.14 ../pom.xml diff --git a/app-stream-network/app-stream-network-ws/pom.xml b/app-stream-network/app-stream-network-ws/pom.xml index b045dae..8e6a32a 100644 --- a/app-stream-network/app-stream-network-ws/pom.xml +++ b/app-stream-network/app-stream-network-ws/pom.xml @@ -4,7 +4,7 @@ com.dingtalk.open app-stream-network - 1.3.13 + 1.3.14 ../pom.xml @@ -20,7 +20,7 @@ com.dingtalk.open app-stream-network-api - ${parent.version} + ${project.parent.version} com.alibaba.fastjson2 @@ -30,5 +30,10 @@ io.netty netty-handler-proxy + + junit + junit + test + diff --git a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java index b2fa344..3344472 100644 --- a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java +++ b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandler.java @@ -12,17 +12,14 @@ import io.netty.handler.codec.http.websocketx.PongWebSocketFrame; import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler; import io.netty.handler.timeout.IdleStateEvent; -import io.netty.util.HashedWheelTimer; -import io.netty.util.Timeout; +import java.nio.charset.StandardCharsets; import java.time.Duration; -import java.util.Iterator; -import java.util.Map; -import java.util.Map.Entry; import java.util.UUID; -import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; /** * @author feiyin @@ -31,15 +28,14 @@ public class KeepAliveHandler extends SimpleChannelInboundHandler { private static final InternalLogger LOGGER = InternalLoggerFactory.getLogger(KeepAliveHandler.class); private final Duration timeout; - private final static HashedWheelTimer TIMER = new HashedWheelTimer(); - private Channel channel; - private final Map timeouts; + private volatile Channel channel; private final AtomicBoolean active; + private final AtomicReference pendingPing; public KeepAliveHandler(Duration timeout) { this.timeout = timeout; this.active = new AtomicBoolean(false); - this.timeouts = new ConcurrentHashMap<>(); + this.pendingPing = new AtomicReference<>(); } @@ -52,7 +48,15 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc } if (evt instanceof IdleStateEvent) { - channel.eventLoop().execute(new PingTask()); + // IdleStateHandler starts counting after TCP connect, before WS handshake. + // Never touch the channel field until handshake completes, otherwise NPE: + // connection operation failed (KeepAliveHandler.java:56) + if (active.get()) { + final Channel ch = channel != null ? channel : ctx.channel(); + if (ch != null && ch.isActive()) { + ch.eventLoop().execute(new PingTask(ch)); + } + } } super.userEventTriggered(ctx, evt); } @@ -60,48 +64,87 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc @Override protected void channelRead0(ChannelHandlerContext ctx, PongWebSocketFrame msg) throws Exception { byte[] data = NettyByteBufUtils.getBytes(msg.content()); - Timeout out = timeouts.remove(new String(data)); - if (out != null) { - out.cancel(); + PendingPing ping = pendingPing.get(); + if (ping != null + && ping.seq.equals(new String(data, StandardCharsets.UTF_8)) + && pendingPing.compareAndSet(ping, null)) { + ping.cancelTimeout(); } } @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { + active.set(false); + channel = null; shutdown(); super.channelInactive(ctx); } private void shutdown() { - Iterator> it = timeouts.entrySet().iterator(); - while (it.hasNext()) { - Entry entry = it.next(); - entry.getValue().cancel(); - it.remove(); + PendingPing ping = pendingPing.getAndSet(null); + if (ping != null) { + ping.cancelTimeout(); + } + } + + private static class PendingPing { + private final String seq; + private volatile ScheduledFuture timeout; + + private PendingPing(String seq) { + this.seq = seq; + } + + private void cancelTimeout() { + ScheduledFuture current = timeout; + if (current != null) { + current.cancel(false); + } } } private class PingTask implements Runnable { + private final Channel target; + + private PingTask(Channel target) { + this.target = target; + } + @Override public void run() { - if (!timeouts.isEmpty()) { + if (!active.get() || target == null || !target.isActive()) { return; } final String seq = UUID.randomUUID().toString(); - ByteBuf byteBuf = Unpooled.copiedBuffer(seq.getBytes()); + final PendingPing ping = new PendingPing(seq); + if (!pendingPing.compareAndSet(null, ping)) { + return; + } + + ScheduledFuture pingTimeout = target.eventLoop().schedule(() -> { + if (pendingPing.compareAndSet(ping, null)) { + active.set(false); + LOGGER.warn("[DingTalk] connection ping timeout, channel is closing"); + target.close(); + } + }, timeout.toMillis(), TimeUnit.MILLISECONDS); + ping.timeout = pingTimeout; + + // channelInactive may clear the pending ping while the timeout is being installed. + if (pendingPing.get() != ping) { + pingTimeout.cancel(false); + return; + } + + ByteBuf byteBuf = Unpooled.copiedBuffer(seq.getBytes(StandardCharsets.UTF_8)); PingWebSocketFrame frame = new PingWebSocketFrame(byteBuf); - channel.writeAndFlush(frame).addListener(future -> { - if (future.isSuccess()) { - Timeout pingTimeout = TIMER.newTimeout(timeout -> { - LOGGER.warn("[DingTalk] connection ping timeout, channel is closing"); - timeouts.remove(seq); - channel.close(); - }, timeout.toMillis(), TimeUnit.MILLISECONDS); - timeouts.put(seq, pingTimeout); - } else { - channel.close(); + target.writeAndFlush(frame).addListener(future -> { + if (!future.isSuccess() && pendingPing.compareAndSet(ping, null)) { + active.set(false); + pingTimeout.cancel(false); + target.close(); } }); } diff --git a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandler.java b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandler.java index c3a2dbb..e8b62ce 100644 --- a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandler.java +++ b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandler.java @@ -10,13 +10,15 @@ import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakeException; import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler; +import java.io.IOException; + /** * @author feiyin * @date 2023/3/27 */ public class NettyClientHandler extends SimpleChannelInboundHandler { - private static final InternalLogger LOGGER = InternalLoggerFactory.getLogger(ProtocolFrameHandler.class); + private static final InternalLogger LOGGER = InternalLoggerFactory.getLogger(NettyClientHandler.class); private final ClientConnectionListener listener; private final String connectionId; @@ -50,8 +52,15 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { if (cause instanceof WebSocketClientHandshakeException) { LOGGER.error("[DingTalk] connection establish error, please check connectionId={}", connectionId, cause); + } else if (cause instanceof IOException) { + // TCP resets and other channel I/O failures are recoverable. The + // channel is closed below and the session pool establishes a + // replacement connection on its next maintenance run. + LOGGER.warn("[DingTalk] connection I/O failed, connectionId={}", connectionId, cause); } else { LOGGER.error("[DingTalk] connection operation failed, connectionId={}", connectionId, cause); } + // Close the channel so session pool can reconnect instead of leaving a half-dead connection. + ctx.close(); } } diff --git a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandler.java b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandler.java index 5b7019e..a719697 100644 --- a/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandler.java +++ b/app-stream-network/app-stream-network-ws/src/main/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandler.java @@ -5,11 +5,11 @@ import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelPromise; import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler; -import io.netty.util.HashedWheelTimer; -import io.netty.util.Timeout; import java.net.SocketAddress; +import java.nio.channels.ClosedChannelException; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -18,9 +18,7 @@ * @date 2023/9/7 */ public class ProtocolConnectHandler extends ChannelDuplexHandler { - private static final HashedWheelTimer TIMER = new HashedWheelTimer(); - - private Timeout timeout; + private ScheduledFuture timeout; private final CompletableFuture future; /** @@ -36,10 +34,11 @@ public ProtocolConnectHandler(CompletableFuture future, Long connectTim @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt == WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE) { - if (timeout != null && !timeout.isExpired()) { - timeout.cancel(); + if (future.complete(ctx.channel())) { + if (timeout != null) { + timeout.cancel(false); + } //执行回调 - future.complete(ctx.channel()); ctx.pipeline().remove(ProtocolConnectHandler.class); } } @@ -48,9 +47,30 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc @Override public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) throws Exception { - this.timeout = TIMER.newTimeout(t -> { - ProtocolConnectHandler.this.future.completeExceptionally(new TimeoutException("connect timeout")); + this.timeout = ctx.executor().schedule(() -> { + if (ProtocolConnectHandler.this.future.completeExceptionally(new TimeoutException("connect timeout"))) { + ctx.close(); + } }, connectTimeout, TimeUnit.MILLISECONDS); super.connect(ctx, remoteAddress, localAddress, promise); + promise.addListener(connectFuture -> { + if (!connectFuture.isSuccess()) { + Throwable cause = connectFuture.cause() != null + ? connectFuture.cause() : new ClosedChannelException(); + if (future.completeExceptionally(cause)) { + timeout.cancel(false); + ctx.close(); + } + } + }); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + future.completeExceptionally(new ClosedChannelException()); + if (timeout != null) { + timeout.cancel(false); + } + super.channelInactive(ctx); } } diff --git a/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java new file mode 100644 index 0000000..fe3978f --- /dev/null +++ b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/KeepAliveHandlerTest.java @@ -0,0 +1,121 @@ +package com.dingtalk.open.app.stream.network.ws; + +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelOutboundHandlerAdapter; +import io.netty.channel.ChannelPromise; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; +import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler; +import io.netty.handler.timeout.IdleStateEvent; +import io.netty.util.ReferenceCountUtil; +import org.junit.Assert; +import org.junit.Test; + +import java.io.IOException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * Regression for Aone 84622857 / customer logs: + * IdleStateEvent before WS handshake must not NPE in KeepAliveHandler. + */ +public class KeepAliveHandlerTest { + + @Test + public void idleBeforeHandshakeDoesNotThrow() { + KeepAliveHandler handler = new KeepAliveHandler(Duration.ofSeconds(5)); + EmbeddedChannel channel = new EmbeddedChannel(handler); + try { + // Multiple idle events before handshake — reproduces Aone 84622857 NPE path. + channel.pipeline().fireUserEventTriggered(IdleStateEvent.FIRST_READER_IDLE_STATE_EVENT); + channel.pipeline().fireUserEventTriggered(IdleStateEvent.READER_IDLE_STATE_EVENT); + channel.pipeline().fireUserEventTriggered(IdleStateEvent.WRITER_IDLE_STATE_EVENT); + Assert.assertTrue(channel.isActive()); + } finally { + channel.finishAndReleaseAll(); + } + } + + @Test + public void idleAfterHandshakeSendsPing() { + KeepAliveHandler handler = new KeepAliveHandler(Duration.ofSeconds(5)); + EmbeddedChannel channel = new EmbeddedChannel(handler); + Object outbound = null; + try { + channel.pipeline().fireUserEventTriggered( + WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE); + channel.pipeline().fireUserEventTriggered(IdleStateEvent.FIRST_READER_IDLE_STATE_EVENT); + channel.runPendingTasks(); + outbound = channel.readOutbound(); + Assert.assertTrue(outbound instanceof PingWebSocketFrame); + Assert.assertNull(channel.readOutbound()); + Assert.assertTrue(channel.isActive()); + } finally { + ReferenceCountUtil.release(outbound); + channel.finishAndReleaseAll(); + } + } + + @Test + public void multipleIdleEventsQueueOnlyOnePingWhileWriteIsPending() { + DelayedWriteHandler delayedWrite = new DelayedWriteHandler(); + KeepAliveHandler handler = new KeepAliveHandler(Duration.ofSeconds(5)); + EmbeddedChannel channel = new EmbeddedChannel(delayedWrite, handler); + try { + channel.pipeline().fireUserEventTriggered( + WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE); + channel.pipeline().fireUserEventTriggered(IdleStateEvent.READER_IDLE_STATE_EVENT); + channel.pipeline().fireUserEventTriggered(IdleStateEvent.WRITER_IDLE_STATE_EVENT); + channel.pipeline().fireUserEventTriggered(IdleStateEvent.ALL_IDLE_STATE_EVENT); + channel.runPendingTasks(); + + Assert.assertEquals("only one ping may be in flight", 1, delayedWrite.writes.get()); + } finally { + delayedWrite.failPending(); + channel.finishAndReleaseAll(); + } + } + + @Test + public void pendingPingWriteStillTimesOut() throws Exception { + DelayedWriteHandler delayedWrite = new DelayedWriteHandler(); + KeepAliveHandler handler = new KeepAliveHandler(Duration.ofMillis(10)); + EmbeddedChannel channel = new EmbeddedChannel(delayedWrite, handler); + try { + channel.pipeline().fireUserEventTriggered( + WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE); + channel.pipeline().fireUserEventTriggered(IdleStateEvent.READER_IDLE_STATE_EVENT); + channel.runPendingTasks(); + + long deadline = System.nanoTime() + Duration.ofSeconds(2).toNanos(); + while (channel.isActive() && System.nanoTime() < deadline) { + Thread.sleep(10); + channel.runPendingTasks(); + } + Assert.assertFalse("pending ping write must be bounded by keepAlive timeout", channel.isActive()); + } finally { + delayedWrite.failPending(); + channel.finishAndReleaseAll(); + } + } + + private static class DelayedWriteHandler extends ChannelOutboundHandlerAdapter { + private final AtomicInteger writes = new AtomicInteger(); + private final List promises = new ArrayList<>(); + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { + writes.incrementAndGet(); + ReferenceCountUtil.release(msg); + promises.add(promise); + } + + private void failPending() { + for (ChannelPromise promise : promises) { + promise.tryFailure(new IOException("test cleanup")); + } + } + } +} diff --git a/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandlerTest.java b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandlerTest.java new file mode 100644 index 0000000..4b0540c --- /dev/null +++ b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/NettyClientHandlerTest.java @@ -0,0 +1,50 @@ +package com.dingtalk.open.app.stream.network.ws; + +import com.dingtalk.open.app.stream.network.api.ClientConnectionListener; +import com.dingtalk.open.app.stream.network.api.Context; +import io.netty.channel.embedded.EmbeddedChannel; +import org.junit.Assert; +import org.junit.Test; + +import java.io.IOException; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * Ensure exceptionCaught closes the channel so session pool can reconnect. + */ +public class NettyClientHandlerTest { + + @Test + public void exceptionCaughtClosesActiveChannel() { + assertExceptionClosesChannel(new RuntimeException("simulated handler failure")); + } + + @Test + public void connectionResetClosesActiveChannel() { + assertExceptionClosesChannel(new IOException("Connection reset by peer")); + } + + private void assertExceptionClosesChannel(Throwable failure) { + final AtomicBoolean disconnected = new AtomicBoolean(false); + NettyClientHandler handler = new NettyClientHandler("conn-test", new ClientConnectionListener() { + @Override + public void receive(Context context) { + } + + @Override + public void onDisConnection(String connectionId) { + disconnected.set(true); + } + }); + + EmbeddedChannel channel = new EmbeddedChannel(handler); + Assert.assertTrue(channel.isActive()); + channel.pipeline().fireExceptionCaught(failure); + + // EmbeddedChannel closes synchronously on close() + Assert.assertFalse("channel should be closed after exceptionCaught", channel.isActive()); + // channelInactive should notify listener + Assert.assertTrue("onDisConnection should be triggered", disconnected.get()); + channel.finishAndReleaseAll(); + } +} diff --git a/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandlerTest.java b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandlerTest.java new file mode 100644 index 0000000..d628d8d --- /dev/null +++ b/app-stream-network/app-stream-network-ws/src/test/java/com/dingtalk/open/app/stream/network/ws/ProtocolConnectHandlerTest.java @@ -0,0 +1,89 @@ +package com.dingtalk.open.app.stream.network.ws; + +import io.netty.channel.Channel; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelOutboundHandlerAdapter; +import io.netty.channel.ChannelPromise; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler; +import org.junit.Assert; +import org.junit.Test; + +import java.net.ConnectException; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +public class ProtocolConnectHandlerTest { + + @Test + public void connectTimeoutClosesChannel() throws Exception { + CompletableFuture future = new CompletableFuture<>(); + EmbeddedChannel channel = new EmbeddedChannel(new ProtocolConnectHandler(future, 10L)); + try { + channel.connect(new InetSocketAddress("localhost", 1234)).syncUninterruptibly(); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(1); + while (!future.isDone() && System.nanoTime() < deadline) { + Thread.sleep(10); + channel.runScheduledPendingTasks(); + } + try { + future.get(1, TimeUnit.SECONDS); + Assert.fail("connect future should time out"); + } catch (ExecutionException expected) { + // expected + } + channel.runPendingTasks(); + Assert.assertFalse("timed-out connection must be closed", channel.isActive()); + } finally { + channel.finishAndReleaseAll(); + } + } + + @Test + public void completedHandshakeIsNotClosedByTimeout() throws Exception { + CompletableFuture future = new CompletableFuture<>(); + EmbeddedChannel channel = new EmbeddedChannel(new ProtocolConnectHandler(future, 10L)); + try { + channel.connect(new InetSocketAddress("localhost", 1234)).syncUninterruptibly(); + channel.pipeline().fireUserEventTriggered( + WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE); + + Assert.assertSame(channel, future.get(1, TimeUnit.SECONDS)); + Thread.sleep(200); + channel.runPendingTasks(); + Assert.assertTrue("completed handshake must win over timeout", channel.isActive()); + } finally { + channel.finishAndReleaseAll(); + } + } + + @Test + public void connectionFailureCompletesFutureAndClosesChannel() throws Exception { + CompletableFuture future = new CompletableFuture<>(); + ChannelOutboundHandlerAdapter failingConnect = new ChannelOutboundHandlerAdapter() { + @Override + public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, + SocketAddress localAddress, ChannelPromise promise) { + promise.setFailure(new ConnectException("simulated connect failure")); + } + }; + EmbeddedChannel channel = new EmbeddedChannel( + failingConnect, new ProtocolConnectHandler(future, 5000L)); + try { + channel.connect(new InetSocketAddress("localhost", 1234)); + try { + future.get(1, TimeUnit.SECONDS); + Assert.fail("connect future should fail"); + } catch (ExecutionException expected) { + Assert.assertTrue(expected.getCause() instanceof ConnectException); + } + channel.runPendingTasks(); + Assert.assertFalse("failed connection must be closed", channel.isActive()); + } finally { + channel.finishAndReleaseAll(); + } + } +} diff --git a/app-stream-network/pom.xml b/app-stream-network/pom.xml index c646a6c..7c66964 100644 --- a/app-stream-network/pom.xml +++ b/app-stream-network/pom.xml @@ -5,7 +5,7 @@ open-app-stream-client com.dingtalk.open - 1.3.13 + 1.3.14 ../pom.xml pom diff --git a/app-stream-protocol/pom.xml b/app-stream-protocol/pom.xml index b88623b..4774766 100644 --- a/app-stream-protocol/pom.xml +++ b/app-stream-protocol/pom.xml @@ -6,7 +6,7 @@ com.dingtalk.open open-app-stream-client - 1.3.13 + 1.3.14 app-stream-protocol diff --git a/dingtalk-stream/pom.xml b/dingtalk-stream/pom.xml index 0d1adc0..385b4eb 100644 --- a/dingtalk-stream/pom.xml +++ b/dingtalk-stream/pom.xml @@ -4,13 +4,13 @@ com.dingtalk.open open-app-stream-client - 1.3.13 + 1.3.14 ../pom.xml dingtalk-stream jar - 1.3.13 + 1.3.14 app-stream-client @@ -78,7 +78,7 @@ com.dingtalk.open:app-stream-network-ws com.dingtalk.open:app-stream-protocol io.netty:* - com.alibaba.fastjson2:* + com.alibaba.fastjson2:* @@ -101,11 +101,7 @@ - - META-INF/services/com.dingtalk.open.app.stream.network.rsocket.RsocketTransportConnector - - + implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/> diff --git a/pom.xml b/pom.xml index 97eceeb..fd37737 100644 --- a/pom.xml +++ b/pom.xml @@ -5,7 +5,7 @@ com.dingtalk.open open-app-stream-client pom - 1.3.13 + 1.3.14 app-stream-client app-stream-api @@ -171,6 +171,13 @@ + + true true