This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new c5e3100 feat: add producer connection endpoint (#510)
c5e3100 is described below
commit c5e310099259913d150bb13f303eba782be5fbc9
Author: aias00 <[email protected]>
AuthorDate: Thu Jul 23 22:43:18 2026 -0700
feat: add producer connection endpoint (#510)
Add producer connection query endpoint with topic/producerGroup filtering,
returning backward-compatible connectionSet format.
---
.../studio/cluster/client/ClientConnectionVO.java | 1 +
.../studio/cluster/client/ClientProviderStub.java | 1 +
...tionVO.java => ProducerConnectionResultVO.java} | 19 +---
.../cluster/client/ProducerConnectionService.java | 63 ++++++++++++
...ConnectionVO.java => ProducerConnectionVO.java} | 18 +---
...ntConnectionVO.java => ProducerController.java} | 39 ++++----
.../client/ProducerConnectionServiceTest.java | 111 +++++++++++++++++++++
.../cluster/client/ProducerControllerTest.java | 67 +++++++++++++
8 files changed, 267 insertions(+), 52 deletions(-)
diff --git
a/server/src/main/java/com/rocketmq/studio/cluster/client/ClientConnectionVO.java
b/server/src/main/java/com/rocketmq/studio/cluster/client/ClientConnectionVO.java
index b9ef5a3..537b7e5 100644
---
a/server/src/main/java/com/rocketmq/studio/cluster/client/ClientConnectionVO.java
+++
b/server/src/main/java/com/rocketmq/studio/cluster/client/ClientConnectionVO.java
@@ -34,6 +34,7 @@ public class ClientConnectionVO {
private String clientId;
private ClientType type;
private String groupOrTopic;
+ private String producerGroup;
private Protocol protocol;
private String address;
private ClientLanguage language;
diff --git
a/server/src/main/java/com/rocketmq/studio/cluster/client/ClientProviderStub.java
b/server/src/main/java/com/rocketmq/studio/cluster/client/ClientProviderStub.java
index 27aa848..7b22d47 100644
---
a/server/src/main/java/com/rocketmq/studio/cluster/client/ClientProviderStub.java
+++
b/server/src/main/java/com/rocketmq/studio/cluster/client/ClientProviderStub.java
@@ -36,6 +36,7 @@ public class ClientProviderStub implements ClientProvider {
.clientId("producer-001")
.type(ClientType.Producer)
.groupOrTopic("order-topic")
+ .producerGroup("pg-order")
.protocol(Protocol.gRPC)
.address("192.168.1.10:56789")
.language(ClientLanguage.Java)
diff --git
a/server/src/main/java/com/rocketmq/studio/cluster/client/ClientConnectionVO.java
b/server/src/main/java/com/rocketmq/studio/cluster/client/ProducerConnectionResultVO.java
similarity index 63%
copy from
server/src/main/java/com/rocketmq/studio/cluster/client/ClientConnectionVO.java
copy to
server/src/main/java/com/rocketmq/studio/cluster/client/ProducerConnectionResultVO.java
index b9ef5a3..9c6934f 100644
---
a/server/src/main/java/com/rocketmq/studio/cluster/client/ClientConnectionVO.java
+++
b/server/src/main/java/com/rocketmq/studio/cluster/client/ProducerConnectionResultVO.java
@@ -16,28 +16,15 @@
*/
package com.rocketmq.studio.cluster.client;
-import com.rocketmq.studio.common.domain.enums.ClientLanguage;
-import com.rocketmq.studio.common.domain.enums.ClientType;
-import com.rocketmq.studio.common.domain.enums.Protocol;
import lombok.AllArgsConstructor;
-import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
-import java.time.LocalDateTime;
+import java.util.List;
@Data
-@Builder
@NoArgsConstructor
@AllArgsConstructor
-public class ClientConnectionVO {
- private String clientId;
- private ClientType type;
- private String groupOrTopic;
- private Protocol protocol;
- private String address;
- private ClientLanguage language;
- private String version;
- private LocalDateTime connectedAt;
- private String clusterName;
+public class ProducerConnectionResultVO {
+ private List<ProducerConnectionVO> connectionSet;
}
diff --git
a/server/src/main/java/com/rocketmq/studio/cluster/client/ProducerConnectionService.java
b/server/src/main/java/com/rocketmq/studio/cluster/client/ProducerConnectionService.java
new file mode 100644
index 0000000..b5d2b77
--- /dev/null
+++
b/server/src/main/java/com/rocketmq/studio/cluster/client/ProducerConnectionService.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.rocketmq.studio.cluster.client;
+
+import com.rocketmq.studio.common.domain.enums.ClientType;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class ProducerConnectionService {
+
+ private final ClientService clientService;
+
+ public List<ProducerConnectionVO> listConnections(String topic, String
producerGroup) {
+ log.info("Listing producer connections, topic={}, producerGroup={}",
topic, producerGroup);
+ return clientService.listConnections(null,
ClientType.Producer.name()).stream()
+ .filter(connection -> matchesFilter(connection, topic,
producerGroup))
+ .map(this::toProducerConnection)
+ .toList();
+ }
+
+ private boolean matchesFilter(ClientConnectionVO connection, String topic,
String producerGroup) {
+ if (hasText(topic) && !topic.equals(connection.getGroupOrTopic())) {
+ return false;
+ }
+ if (hasText(producerGroup) &&
!producerGroup.equals(connection.getProducerGroup())) {
+ return false;
+ }
+ return true;
+ }
+
+ private ProducerConnectionVO toProducerConnection(ClientConnectionVO
connection) {
+ return ProducerConnectionVO.builder()
+ .clientId(connection.getClientId())
+ .clientAddr(connection.getAddress())
+ .language(connection.getLanguage() == null ? null :
connection.getLanguage().name())
+ .versionDesc(connection.getVersion())
+ .build();
+ }
+
+ private boolean hasText(String value) {
+ return value != null && !value.trim().isEmpty();
+ }
+}
diff --git
a/server/src/main/java/com/rocketmq/studio/cluster/client/ClientConnectionVO.java
b/server/src/main/java/com/rocketmq/studio/cluster/client/ProducerConnectionVO.java
similarity index 67%
copy from
server/src/main/java/com/rocketmq/studio/cluster/client/ClientConnectionVO.java
copy to
server/src/main/java/com/rocketmq/studio/cluster/client/ProducerConnectionVO.java
index b9ef5a3..aa4c213 100644
---
a/server/src/main/java/com/rocketmq/studio/cluster/client/ClientConnectionVO.java
+++
b/server/src/main/java/com/rocketmq/studio/cluster/client/ProducerConnectionVO.java
@@ -16,28 +16,18 @@
*/
package com.rocketmq.studio.cluster.client;
-import com.rocketmq.studio.common.domain.enums.ClientLanguage;
-import com.rocketmq.studio.common.domain.enums.ClientType;
-import com.rocketmq.studio.common.domain.enums.Protocol;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
-import java.time.LocalDateTime;
-
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
-public class ClientConnectionVO {
+public class ProducerConnectionVO {
private String clientId;
- private ClientType type;
- private String groupOrTopic;
- private Protocol protocol;
- private String address;
- private ClientLanguage language;
- private String version;
- private LocalDateTime connectedAt;
- private String clusterName;
+ private String clientAddr;
+ private String language;
+ private String versionDesc;
}
diff --git
a/server/src/main/java/com/rocketmq/studio/cluster/client/ClientConnectionVO.java
b/server/src/main/java/com/rocketmq/studio/cluster/client/ProducerController.java
similarity index 51%
copy from
server/src/main/java/com/rocketmq/studio/cluster/client/ClientConnectionVO.java
copy to
server/src/main/java/com/rocketmq/studio/cluster/client/ProducerController.java
index b9ef5a3..32d9611 100644
---
a/server/src/main/java/com/rocketmq/studio/cluster/client/ClientConnectionVO.java
+++
b/server/src/main/java/com/rocketmq/studio/cluster/client/ProducerController.java
@@ -16,28 +16,23 @@
*/
package com.rocketmq.studio.cluster.client;
-import com.rocketmq.studio.common.domain.enums.ClientLanguage;
-import com.rocketmq.studio.common.domain.enums.ClientType;
-import com.rocketmq.studio.common.domain.enums.Protocol;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Data;
-import lombok.NoArgsConstructor;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
-import java.time.LocalDateTime;
+@RestController
+@RequestMapping("/api/producer")
+@RequiredArgsConstructor
+public class ProducerController {
-@Data
-@Builder
-@NoArgsConstructor
-@AllArgsConstructor
-public class ClientConnectionVO {
- private String clientId;
- private ClientType type;
- private String groupOrTopic;
- private Protocol protocol;
- private String address;
- private ClientLanguage language;
- private String version;
- private LocalDateTime connectedAt;
- private String clusterName;
+ private final ProducerConnectionService producerConnectionService;
+
+ @GetMapping("/connection")
+ public ProducerConnectionResultVO listConnections(
+ @RequestParam(required = false) String topic,
+ @RequestParam(required = false) String producerGroup) {
+ return new
ProducerConnectionResultVO(producerConnectionService.listConnections(topic,
producerGroup));
+ }
}
diff --git
a/server/src/test/java/com/rocketmq/studio/cluster/client/ProducerConnectionServiceTest.java
b/server/src/test/java/com/rocketmq/studio/cluster/client/ProducerConnectionServiceTest.java
new file mode 100644
index 0000000..c3d700f
--- /dev/null
+++
b/server/src/test/java/com/rocketmq/studio/cluster/client/ProducerConnectionServiceTest.java
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.rocketmq.studio.cluster.client;
+
+import com.rocketmq.studio.common.domain.enums.ClientLanguage;
+import com.rocketmq.studio.common.domain.enums.ClientType;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class ProducerConnectionServiceTest {
+
+ @Mock
+ private ClientService clientService;
+
+ @InjectMocks
+ private ProducerConnectionService producerConnectionService;
+
+ @Test
+ void listConnectionsShouldProjectProducerClientsByTopic() {
+ ClientConnectionVO producer = ClientConnectionVO.builder()
+ .clientId("producer-1")
+ .type(ClientType.Producer)
+ .groupOrTopic("order-topic")
+ .producerGroup("pg-order")
+ .address("10.0.0.1:38888")
+ .language(ClientLanguage.Java)
+ .version("5.1.0")
+ .build();
+ ClientConnectionVO otherProducer = ClientConnectionVO.builder()
+ .clientId("producer-2")
+ .type(ClientType.Producer)
+ .groupOrTopic("payment-topic")
+ .producerGroup("pg-payment")
+ .address("10.0.0.2:38888")
+ .language(ClientLanguage.Go)
+ .version("5.0.0")
+ .build();
+ when(clientService.listConnections(null, ClientType.Producer.name()))
+ .thenReturn(List.of(producer, otherProducer));
+
+ List<ProducerConnectionVO> result =
producerConnectionService.listConnections("order-topic", "pg-order");
+
+ assertThat(result).hasSize(1);
+ assertThat(result.get(0).getClientId()).isEqualTo("producer-1");
+ assertThat(result.get(0).getClientAddr()).isEqualTo("10.0.0.1:38888");
+ assertThat(result.get(0).getLanguage()).isEqualTo("Java");
+ assertThat(result.get(0).getVersionDesc()).isEqualTo("5.1.0");
+ verify(clientService).listConnections(null,
ClientType.Producer.name());
+ }
+
+ @Test
+ void listConnectionsShouldFallbackToProducerGroupWhenTopicIsMissing() {
+ ClientConnectionVO producer = ClientConnectionVO.builder()
+ .clientId("producer-1")
+ .type(ClientType.Producer)
+ .groupOrTopic("order-topic")
+ .producerGroup("pg-order")
+ .address("10.0.0.1:38888")
+ .language(ClientLanguage.Java)
+ .version("5.1.0")
+ .build();
+ when(clientService.listConnections(null,
ClientType.Producer.name())).thenReturn(List.of(producer));
+
+ List<ProducerConnectionVO> result =
producerConnectionService.listConnections(null, "pg-order");
+
+ assertThat(result).hasSize(1);
+ assertThat(result.get(0).getClientId()).isEqualTo("producer-1");
+ }
+
+ @Test
+ void listConnectionsShouldRequireProducerGroupWhenBothFiltersAreProvided()
{
+ ClientConnectionVO producer = ClientConnectionVO.builder()
+ .clientId("producer-1")
+ .type(ClientType.Producer)
+ .groupOrTopic("order-topic")
+ .producerGroup("pg-order")
+ .address("10.0.0.1:38888")
+ .language(ClientLanguage.Java)
+ .version("5.1.0")
+ .build();
+ when(clientService.listConnections(null,
ClientType.Producer.name())).thenReturn(List.of(producer));
+
+ List<ProducerConnectionVO> result =
producerConnectionService.listConnections("order-topic", "wrong-group");
+
+ assertThat(result).isEmpty();
+ }
+}
diff --git
a/server/src/test/java/com/rocketmq/studio/cluster/client/ProducerControllerTest.java
b/server/src/test/java/com/rocketmq/studio/cluster/client/ProducerControllerTest.java
new file mode 100644
index 0000000..c05cf3b
--- /dev/null
+++
b/server/src/test/java/com/rocketmq/studio/cluster/client/ProducerControllerTest.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.rocketmq.studio.cluster.client;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import
org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
+import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.test.web.servlet.MockMvc;
+
+import java.util.List;
+
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static
org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+@WebMvcTest(ProducerController.class)
+@AutoConfigureMockMvc(addFilters = false)
+class ProducerControllerTest {
+
+ @Autowired
+ private MockMvc mockMvc;
+
+ @MockBean
+ private ProducerConnectionService producerConnectionService;
+
+ @Test
+ void listConnectionsShouldReturnLegacyConnectionSetPayload() throws
Exception {
+ ProducerConnectionVO connection = ProducerConnectionVO.builder()
+ .clientId("producer-1")
+ .clientAddr("10.0.0.1:38888")
+ .language("Java")
+ .versionDesc("5.1.0")
+ .build();
+ when(producerConnectionService.listConnections("order-topic",
"pg-order"))
+ .thenReturn(List.of(connection));
+
+ mockMvc.perform(get("/api/producer/connection")
+ .param("topic", "order-topic")
+ .param("producerGroup", "pg-order"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.connectionSet").isArray())
+
.andExpect(jsonPath("$.connectionSet[0].clientId").value("producer-1"))
+
.andExpect(jsonPath("$.connectionSet[0].clientAddr").value("10.0.0.1:38888"))
+
.andExpect(jsonPath("$.connectionSet[0].language").value("Java"))
+
.andExpect(jsonPath("$.connectionSet[0].versionDesc").value("5.1.0"));
+
+ verify(producerConnectionService).listConnections("order-topic",
"pg-order");
+ }
+}
