fix: avoid KeepAlive NPE when idle fires before WS handshake

.github/workflows/release.yml

@@ -16,17 +16,17 @@ 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
 
       - 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

.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

app-stream-api/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>open-app-stream-client</artifactId>
         <groupId>com.dingtalk.open</groupId>
-        <version>1.3.13</version>
+        <version>1.3.14</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

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,45 +42,99 @@ class OpenDingTalkStreamClient implements OpenDingTalkClient {
     private NetWorkService netWorkService;
     private OpenApiClient openApiClient;
     private Set<Subscription> subscriptions;
+    private boolean networkResourcesAcquired;
 
     public OpenDingTalkStreamClient(DingTalkCredential credential, CommandDispatcher dispatcher, ExecutorService executor, ClientOption option, Set<Subscription> subscriptions,
                                     NetProxy netProxy) {
         this.credential = credential;
         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;
     }
 
     @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);
         }
     }
 
     @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<Subscription> subscriptions, NetProxy proxy) throws Exception {
         OpenConnectionRequest request = new OpenConnectionRequest();
         request.setClientId(credential.getClientId());

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,14 +154,16 @@ public OpenDingTalkStreamClientBuilder forwardGraphRequestToHTTP(int port) {
 
 
     public OpenDingTalkClient build() {
+        DingTalkCredential validCredential = Preconditions.notNull(credential);
         ClientOption option = new ClientOption();
         option.setConnectTimeout(connectTimeout);
         option.setMaxConnectionCount(maxConnectionCount);
         option.setConnectionTTL(connectionTimeToLive);
         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) {

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<CommandType, CommandExecutor> registry;
 
     public CommandDispatcher(Map<CommandType, CommandExecutor> registry) {
-        registry.putIfAbsent(CommandType.SYSTEM, new SystemCommandExecutor());
-        this.registry = Collections.unmodifiableMap(registry);
+        Map<CommandType, CommandExecutor> registrySnapshot = new HashMap<>(registry);
+        registrySnapshot.putIfAbsent(CommandType.SYSTEM, new SystemCommandExecutor());
+        this.registry = Collections.unmodifiableMap(registrySnapshot);
     }
 
     /**