refactor(proxy): 重构证书变更事件处理逻辑

EnableAsync · EnableAsync · commit 611a38dd2fe4 · 2025-06-19T15:31:55.000+08:00
- 修改 CertChangeEvent 类，将 values 字段改为单个 value 字段
- 更新 CertChangeSource 接口，使用事件列表替代单个事件
-调整 FileCertChangeSource 实现，适应新的事件处理方式- 修改 TlsCertificateManager 类，支持处理多个证书变更事件
- 更新相关测试用例，以适应新的事件处理逻辑

Signed-off-by: Async &lt;raisinata@foxmail.com&gt;
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/ProxyAndTlsProtocolNegotiator.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/ProxyAndTlsProtocolNegotiator.java
@@ -134,7 +134,7 @@ public static void loadSslContext(InputStream certInputStream,
                 String tlsKeyPath = ConfigurationManager.getProxyConfig().getTlsKeyPath();
                 try (InputStream serverKeyInputStream = Files.newInputStream(
                     Paths.get(tlsKeyPath));
-                     InputStream serverCertificateStream = Files.newInputStream(
+                    InputStream serverCertificateStream = Files.newInputStream(
                          Paths.get(tlsCertPath))) {
                     sslContext = GrpcSslContexts.forServer(serverCertificateStream,
                             serverKeyInputStream)
@@ -166,8 +166,8 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
                 }
                 if (ha.state() == ProtocolDetectionState.DETECTED) {
                     ctx.pipeline().addAfter(ctx.name(), HA_PROXY_DECODER, new HAProxyMessageDecoder())
-                            .addAfter(HA_PROXY_DECODER, HA_PROXY_HANDLER, new HAProxyMessageHandler())
-                            .addAfter(HA_PROXY_HANDLER, TLS_MODE_HANDLER, new TlsModeHandler(grpcHandler));
+                        .addAfter(HA_PROXY_DECODER, HA_PROXY_HANDLER, new HAProxyMessageHandler())
+                        .addAfter(HA_PROXY_HANDLER, TLS_MODE_HANDLER, new TlsModeHandler(grpcHandler));
                 } else {
                     ctx.pipeline().addAfter(ctx.name(), TLS_MODE_HANDLER, new TlsModeHandler(grpcHandler));
                 }
@@ -233,7 +233,7 @@ private void handleWithMessage(HAProxyMessage msg) {
                     msg.tlvs().forEach(tlv -> handleHAProxyTLV(tlv, builder));
                 }
                 pne = InternalProtocolNegotiationEvent
-                        .withAttributes(InternalProtocolNegotiationEvent.getDefault(), builder.build());
+                    .withAttributes(InternalProtocolNegotiationEvent.getDefault(), builder.build());
             } finally {
                 msg.release();
             }
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/cert/CertChangeEvent.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/cert/CertChangeEvent.java
@@ -16,8 +16,6 @@
  */
 package org.apache.rocketmq.proxy.service.cert;
 
-import java.util.List;
-
 public class CertChangeEvent {
 
     public enum Kind {
@@ -35,12 +33,12 @@ public enum SourceType {
 
     private final SourceType sourceType;
 
-    private final List<String> values;
+    private final String value;
 
-    public CertChangeEvent(Kind kind, SourceType type, List<String> values) {
+    public CertChangeEvent(Kind kind, SourceType type, String value) {
         this.kind = kind;
         this.sourceType = type;
-        this.values = values;
+        this.value = value;
     }
 
     public Kind getKind() {
@@ -51,15 +49,15 @@ public SourceType getSourceType() {
         return sourceType;
     }
 
-    public List<String> getValues() {
-        return values;
+    public String getValue() {
+        return value;
     }
 
     @Override public String toString() {
         return "CertChangeEvent{" +
             "kind=" + kind +
             ", sourceType=" + sourceType +
-            ", values=" + values +
+            ", values=" + value +
             '}';
     }
 }
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/cert/CertChangeSource.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/cert/CertChangeSource.java
@@ -16,13 +16,14 @@
  */
 package org.apache.rocketmq.proxy.service.cert;
 
+import java.util.List;
 import org.apache.rocketmq.common.utils.StartAndShutdown;
 
 public interface CertChangeSource extends StartAndShutdown {
 
     void setListener(ChangeListener listener);
 
     interface ChangeListener {
-        void onCertChanged(CertChangeEvent event);
+        void onCertChanged(List<CertChangeEvent> events);
     }
 }
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/cert/FileCertChangeSource.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/cert/FileCertChangeSource.java
@@ -48,8 +48,7 @@ public FileCertChangeSource(String certPath, String keyPath, int watchInterval)
                         kind = CertChangeEvent.Kind.SERVER_KEY;
                     }
 
-                    listener.onCertChanged(new CertChangeEvent(kind, CertChangeEvent.SourceType.FILE,
-                        Collections.singletonList(path)));
+                    listener.onCertChanged(Collections.singletonList(new CertChangeEvent(kind, CertChangeEvent.SourceType.FILE, path)));
                 }, watchInterval);
         } catch (Exception e) {
             throw new RuntimeException(e);
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/cert/TlsCertificateManager.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/cert/TlsCertificateManager.java
@@ -66,14 +66,15 @@ public void shutdown() throws Exception {
     }
 
     @Override
-    public void onCertChanged(CertChangeEvent event) {
-        log.info("cert changed: {}", event);
-        if (event.getSourceType() == CertChangeEvent.SourceType.FILE) {
-            if (event.getValues().isEmpty()) {
+    public void onCertChanged(List<CertChangeEvent> events) {
+        log.info("cert changed: {}", events);
+        if (events.size() == 1 && events.get(0).getSourceType() == CertChangeEvent.SourceType.FILE) {
+            CertChangeEvent event = events.get(0);
+            if (event.getValue() == null) {
                 log.warn("cert path is empty, ignore");
                 return;
             }
-            String path = event.getValues().get(0);
+            String path = event.getValue();
             if (path.equals(ConfigurationManager.getProxyConfig().getTlsCertPath())) {
                 certChanged = true;
             } else if (path.equals(ConfigurationManager.getProxyConfig().getTlsKeyPath())) {
@@ -88,15 +89,15 @@ public void onCertChanged(CertChangeEvent event) {
                 certChanged = false;
                 keyChanged = false;
             }
-        } else if (event.getSourceType() == CertChangeEvent.SourceType.INLINE) {
-            if (event.getValues().size() != 2) {
-                log.warn("cert inline is invalid, ignore");
+        } else if (events.size() == 2 && events.get(0).getSourceType() == CertChangeEvent.SourceType.INLINE) {
+            if (events.get(0).getValue() == null || events.get(1).getValue() == null) {
+                log.warn("cert or key value is empty, ignore");
                 return;
             }
             for (TlsContextReloadListener listener : reloadListeners) {
                 listener.onTlsContextReload(
-                    IOUtils.toInputStream(event.getValues().get(0), StandardCharsets.UTF_8),
-                    IOUtils.toInputStream(event.getValues().get(1), StandardCharsets.UTF_8)
+                    IOUtils.toInputStream(events.get(0).getValue(), StandardCharsets.UTF_8),
+                    IOUtils.toInputStream(events.get(1).getValue(), StandardCharsets.UTF_8)
                 );
             }
         }
diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/cert/FileCertChangeSourceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/cert/FileCertChangeSourceTest.java
@@ -18,6 +18,7 @@
 
 import java.io.File;
 import java.io.PrintWriter;
+import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import org.apache.rocketmq.proxy.config.ConfigurationManager;
@@ -120,43 +121,51 @@ public void testCertFileChangeTriggerListener() throws Exception {
         boolean called = latch.await(2, TimeUnit.SECONDS);
         assertThat(called).isTrue();
 
-        ArgumentCaptor<CertChangeEvent> eventCaptor = ArgumentCaptor.forClass(CertChangeEvent.class);
-        verify(mockListener, times(1)).onCertChanged(eventCaptor.capture());
-        CertChangeEvent capturedEvent = eventCaptor.getValue();
+        // Use a List<CertChangeEvent> captor
+        @SuppressWarnings("unchecked")
+        ArgumentCaptor<List<CertChangeEvent>> eventListCaptor = ArgumentCaptor.forClass(List.class);
+        verify(mockListener, times(1)).onCertChanged(eventListCaptor.capture());
+
+        List<CertChangeEvent> capturedEvents = eventListCaptor.getValue();
+        assertThat(capturedEvents).hasSize(1);
+        CertChangeEvent capturedEvent = capturedEvents.get(0);
         assertThat(capturedEvent.getKind()).isEqualTo(CertChangeEvent.Kind.SERVER_CERT);
         assertThat(capturedEvent.getSourceType()).isEqualTo(CertChangeEvent.SourceType.FILE);
-        assertThat(capturedEvent.getValues()).contains(certFile.getAbsolutePath());
+        assertThat(capturedEvent.getValue()).contains(certFile.getAbsolutePath());
     }
 
     @Test
     public void testKeyFileChangeTriggerListener() throws Exception {
         fileCertChangeSourceCustomInterval.start();
-
         final CountDownLatch latch = new CountDownLatch(1);
-
-        CertChangeSource.ChangeListener countingListener = event -> {
+        CertChangeSource.ChangeListener countingListener = events -> {
             try {
-                mockListener.onCertChanged(event);
+                mockListener.onCertChanged(events);
             } finally {
                 latch.countDown();
             }
         };
-
         fileCertChangeSourceCustomInterval.setListener(countingListener);
-
         modifyFile(keyFile, "Modified key content " + System.currentTimeMillis());
-
         boolean called = latch.await(2, TimeUnit.SECONDS);
         assertThat(called).isTrue();
 
-        ArgumentCaptor<CertChangeEvent> eventCaptor = ArgumentCaptor.forClass(CertChangeEvent.class);
-        verify(mockListener, times(1)).onCertChanged(eventCaptor.capture());
-        CertChangeEvent capturedEvent = eventCaptor.getValue();
+        // Use a List<CertChangeEvent> captor
+        @SuppressWarnings("unchecked")
+        ArgumentCaptor<List<CertChangeEvent>> eventListCaptor = ArgumentCaptor.forClass(List.class);
+        verify(mockListener, times(1)).onCertChanged(eventListCaptor.capture());
+
+        List<CertChangeEvent> capturedEvents = eventListCaptor.getValue();
+        assertThat(capturedEvents).hasSize(1);
+        CertChangeEvent capturedEvent = capturedEvents.get(0);
         assertThat(capturedEvent.getKind()).isEqualTo(CertChangeEvent.Kind.SERVER_KEY);
         assertThat(capturedEvent.getSourceType()).isEqualTo(CertChangeEvent.SourceType.FILE);
-        assertThat(capturedEvent.getValues()).contains(keyFile.getAbsolutePath());
+        assertThat(capturedEvent.getValue()).contains(keyFile.getAbsolutePath());
     }
 
+
+
+
     @Test
     public void testFileChangeWithoutListener() throws Exception {
 
@@ -169,7 +178,7 @@ public void testFileChangeWithoutListener() throws Exception {
 
         Thread.sleep(500);
 
-        verify(mockListener, times(0)).onCertChanged(Mockito.any(CertChangeEvent.class));
+        verify(mockListener, times(0)).onCertChanged(Mockito.anyList());
 
         sourceWithoutListener.shutdown();
     }
@@ -204,7 +213,7 @@ public void testSetListener() throws Exception {
         boolean called = latch.await(2, TimeUnit.SECONDS);
         assertThat(called).isTrue();
 
-        verify(mockListener, times(1)).onCertChanged(Mockito.any(CertChangeEvent.class));
+        verify(mockListener, times(1)).onCertChanged(Mockito.anyList());
 
         source.shutdown();
     }
diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/cert/TlsCertificateManagerTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/cert/TlsCertificateManagerTest.java

Original file line number	Diff line number	Diff line change
`@@ -16,13 +16,14 @@`
`16`	`16`	`*/`
`17`	`17`	`package org.apache.rocketmq.proxy.service.cert;`
`18`	`18`
	`19`	`+import java.util.List;`
`19`	`20`	`import org.apache.rocketmq.common.utils.StartAndShutdown;`
`20`	`21`
`21`	`22`	`public interface CertChangeSource extends StartAndShutdown {`
`22`	`23`
`23`	`24`	`void setListener(ChangeListener listener);`
`24`	`25`
`25`	`26`	`interface ChangeListener {`
`26`		`- void onCertChanged(CertChangeEvent event);`
	`27`	`+ void onCertChanged(List<CertChangeEvent> events);`
`27`	`28`	`}`
`28`	`29`	`}`
Original file line number	Diff line number	Diff line change
`@@ -48,8 +48,7 @@ public FileCertChangeSource(String certPath, String keyPath, int watchInterval)`
`48`	`48`	`kind = CertChangeEvent.Kind.SERVER_KEY;`
`49`	`49`	`}`
`50`	`50`
`51`		`- listener.onCertChanged(new CertChangeEvent(kind, CertChangeEvent.SourceType.FILE,`
`52`		`- Collections.singletonList(path)));`
	`51`	`+ listener.onCertChanged(Collections.singletonList(new CertChangeEvent(kind, CertChangeEvent.SourceType.FILE, path)));`
`53`	`52`	`}, watchInterval);`
`54`	`53`	`} catch (Exception e) {`
`55`	`54`	`throw new RuntimeException(e);`