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
commit 524c42002480de1771bed5f7c03d4fe5ec9736a0 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 b9ef5a30..537b7e54 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 27aa8483..7b22d47d 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 b9ef5a30..9c6934f7 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 00000000..b5d2b776 --- /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 b9ef5a30..aa4c2139 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 b9ef5a30..32d9611f 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 00000000..c3d700f1 --- /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 00000000..c05cf3b9 --- /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"); + } +}
