mirror of
https://github.com/apache/rocketmq.git
synced 2026-09-24 16:04:00 +08:00
[ISSUE #3949] v2 support
This commit is contained in:
kaiyi.lk
+2
-1
@@ -278,7 +278,8 @@ public class ConsumerService extends BaseService {
|
||||
String brokerAddr = this.getBrokerAddr(ctx, receiptHandle.getBrokerName());
|
||||
|
||||
ChangeInvisibleTimeRequestHeader requestHeader = convertToChangeInvisibleTimeRequestHeader(ctx, request);
|
||||
future = this.writeConsumer.changeInvisibleTimeAsync(ctx, brokerAddr, receiptHandle.getBrokerName(), "", requestHeader)
|
||||
future = this.writeConsumer.changeInvisibleTimeAsync(ctx, brokerAddr, receiptHandle.getBrokerName(),
|
||||
request.getMessageId(), requestHeader)
|
||||
.thenApply(result -> convertToChangeInvisibleDurationResponse(ctx, request, result));
|
||||
} catch (Throwable t) {
|
||||
future.completeExceptionally(t);
|
||||
|
||||
+12
-7
@@ -61,16 +61,19 @@ public class RouteServiceTest extends BaseServiceTest {
|
||||
.setResourceNamespace(NAMESPACE)
|
||||
.build();
|
||||
|
||||
private static final Settings WITH_HOST_SETTINGS = Settings.newBuilder()
|
||||
.setAccessPoint(Endpoints.newBuilder()
|
||||
.addAddresses(Address.newBuilder()
|
||||
.setPort(80)
|
||||
.setHost("host")
|
||||
.build())
|
||||
.setScheme(AddressScheme.DOMAIN_NAME)
|
||||
private static final Endpoints WITH_HOST_ENDPOINT = Endpoints.newBuilder()
|
||||
.addAddresses(Address.newBuilder()
|
||||
.setPort(80)
|
||||
.setHost("host")
|
||||
.build())
|
||||
.setScheme(AddressScheme.DOMAIN_NAME)
|
||||
.build();
|
||||
|
||||
private static final Settings WITH_HOST_SETTINGS = Settings.newBuilder()
|
||||
.setAccessPoint(WITH_HOST_ENDPOINT)
|
||||
.build();
|
||||
|
||||
|
||||
private static final Settings INVALID_HOST_SETTINGS = Settings.newBuilder()
|
||||
.setAccessPoint(Endpoints.getDefaultInstance())
|
||||
.build();
|
||||
@@ -185,6 +188,7 @@ public class RouteServiceTest extends BaseServiceTest {
|
||||
.setTopic(Resource.newBuilder()
|
||||
.setName("topic")
|
||||
.build())
|
||||
.setEndpoints(WITH_HOST_ENDPOINT)
|
||||
.build());
|
||||
|
||||
QueryRouteResponse response = future.get();
|
||||
@@ -228,6 +232,7 @@ public class RouteServiceTest extends BaseServiceTest {
|
||||
when(grpcClientManager.getClientSettings(any(Context.class))).thenReturn(WITH_HOST_SETTINGS);
|
||||
|
||||
CompletableFuture<QueryAssignmentResponse> future = routeService.queryAssignment(Context.current(), QueryAssignmentRequest.newBuilder()
|
||||
.setEndpoints(WITH_HOST_ENDPOINT)
|
||||
.setTopic(Resource.newBuilder()
|
||||
.setName("topic")
|
||||
.build())
|
||||
|
||||
@@ -1,123 +0,0 @@
|
||||
/*
|
||||
* 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 org.apache.rocketmq.test.grpc.v1;
|
||||
|
||||
import apache.rocketmq.v1.AckMessageResponse;
|
||||
import apache.rocketmq.v1.Address;
|
||||
import apache.rocketmq.v1.AddressScheme;
|
||||
import apache.rocketmq.v1.Endpoints;
|
||||
import apache.rocketmq.v1.MessagingServiceGrpc;
|
||||
import apache.rocketmq.v1.QueryRouteResponse;
|
||||
import apache.rocketmq.v1.ReceiveMessageResponse;
|
||||
import apache.rocketmq.v1.SendMessageResponse;
|
||||
import com.google.common.base.Stopwatch;
|
||||
import io.grpc.Channel;
|
||||
import java.net.URL;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import org.apache.rocketmq.proxy.config.ConfigurationManager;
|
||||
import org.apache.rocketmq.proxy.grpc.v1.GrpcMessagingProcessor;
|
||||
import org.apache.rocketmq.proxy.grpc.v1.service.ClusterGrpcService;
|
||||
import org.apache.rocketmq.proxy.grpc.v1.service.GrpcForwardService;
|
||||
import org.junit.After;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
import static org.apache.rocketmq.common.message.MessageClientIDSetter.createUniqID;
|
||||
import static org.apache.rocketmq.proxy.config.ConfigurationManager.RMQ_PROXY_HOME;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
public class ClusterGrpcTest extends GrpcBaseTest {
|
||||
|
||||
private final int PORT = 8083;
|
||||
private GrpcForwardService grpcForwardService;
|
||||
private MessagingServiceGrpc.MessagingServiceBlockingStub blockingStub;
|
||||
|
||||
@Before
|
||||
public void setUp() throws Exception {
|
||||
String mockProxyHome = "/mock/rmq/proxy/home";
|
||||
URL mockProxyHomeURL = getClass().getClassLoader().getResource("rmq-proxy-home");
|
||||
if (mockProxyHomeURL != null) {
|
||||
mockProxyHome = mockProxyHomeURL.toURI().getPath();
|
||||
}
|
||||
System.setProperty(RMQ_PROXY_HOME, mockProxyHome);
|
||||
ConfigurationManager.initEnv();
|
||||
ConfigurationManager.intConfig();
|
||||
ConfigurationManager.getProxyConfig().setGrpcServerPort(PORT);
|
||||
ConfigurationManager.getProxyConfig().setNameSrvAddr(nsAddr);
|
||||
grpcForwardService = new ClusterGrpcService();
|
||||
grpcForwardService.start();
|
||||
GrpcMessagingProcessor processor = new GrpcMessagingProcessor(grpcForwardService);
|
||||
Channel channel = setUpServer(processor, ConfigurationManager.getProxyConfig().getGrpcServerPort(), true);
|
||||
blockingStub = MessagingServiceGrpc.newBlockingStub(channel);
|
||||
}
|
||||
|
||||
@After
|
||||
public void tearDown() throws Exception {
|
||||
grpcForwardService.shutdown();
|
||||
shutdown();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testQueryRoute() {
|
||||
String topic = initTopic();
|
||||
QueryRouteResponse response = blockingStub.queryRoute(buildQueryRouteRequest(topic, Endpoints.newBuilder()
|
||||
.setScheme(AddressScheme.IPv4)
|
||||
.addAddresses(Address.newBuilder()
|
||||
.setHost("127.0.0.1")
|
||||
.setPort(PORT)
|
||||
.build())
|
||||
.build()));
|
||||
assertQueryRoute(response, brokerControllerList.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSendReceiveMessage() {
|
||||
String group = "group";
|
||||
String messageId = createUniqID();
|
||||
SendMessageResponse sendResponse = blockingStub.sendMessage(buildSendMessageRequest(broker1Name, messageId));
|
||||
assertSendMessage(sendResponse, messageId);
|
||||
|
||||
ReceiveMessageResponse receiveResponse = blockingStub.withDeadlineAfter(3, TimeUnit.SECONDS)
|
||||
.receiveMessage(buildReceiveMessageRequest(group, broker1Name));
|
||||
assertReceiveMessage(receiveResponse, messageId);
|
||||
String receiptHandle = receiveResponse.getMessages(0).getSystemAttribute().getReceiptHandle();
|
||||
AckMessageResponse ackMessageResponse = blockingStub.ackMessage(buildAckMessageRequest(group, broker1Name, receiptHandle));
|
||||
assertAck(ackMessageResponse);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSendReceiveDelayMessage() {
|
||||
String group = "group";
|
||||
String messageId = createUniqID();
|
||||
SendMessageResponse sendResponse = blockingStub.sendMessage(buildSendDelayMessageRequest(broker1Name, messageId, 2));
|
||||
assertSendMessage(sendResponse, messageId);
|
||||
|
||||
Stopwatch stopwatch = Stopwatch.createStarted();
|
||||
ReceiveMessageResponse receiveResponse = blockingStub.withDeadlineAfter(10, TimeUnit.SECONDS)
|
||||
.receiveMessage(buildReceiveMessageRequest(group, broker1Name));
|
||||
long rcvTime = stopwatch.elapsed(TimeUnit.SECONDS);
|
||||
assertTrue(Math.abs(rcvTime - 5) < 2);
|
||||
|
||||
assertReceiveMessage(receiveResponse, messageId);
|
||||
String receiptHandle = receiveResponse.getMessages(0).getSystemAttribute().getReceiptHandle();
|
||||
AckMessageResponse ackMessageResponse = blockingStub.ackMessage(buildAckMessageRequest(group, broker1Name, receiptHandle));
|
||||
assertAck(ackMessageResponse);
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@@ -1,207 +0,0 @@
|
||||
/*
|
||||
* 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 org.apache.rocketmq.test.grpc.v1;
|
||||
|
||||
import apache.rocketmq.v1.AckMessageRequest;
|
||||
import apache.rocketmq.v1.AckMessageResponse;
|
||||
import apache.rocketmq.v1.Endpoints;
|
||||
import apache.rocketmq.v1.Message;
|
||||
import apache.rocketmq.v1.Partition;
|
||||
import apache.rocketmq.v1.QueryRouteRequest;
|
||||
import apache.rocketmq.v1.QueryRouteResponse;
|
||||
import apache.rocketmq.v1.ReceiveMessageRequest;
|
||||
import apache.rocketmq.v1.ReceiveMessageResponse;
|
||||
import apache.rocketmq.v1.Resource;
|
||||
import apache.rocketmq.v1.SendMessageRequest;
|
||||
import apache.rocketmq.v1.SendMessageResponse;
|
||||
import apache.rocketmq.v1.SystemAttribute;
|
||||
import com.google.protobuf.ByteString;
|
||||
import com.google.protobuf.Duration;
|
||||
import com.google.protobuf.Timestamp;
|
||||
import com.google.rpc.Code;
|
||||
import io.grpc.Channel;
|
||||
import io.grpc.ServerInterceptors;
|
||||
import io.grpc.ServerServiceDefinition;
|
||||
import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
|
||||
import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
|
||||
import io.grpc.netty.shaded.io.netty.handler.ssl.ApplicationProtocolConfig;
|
||||
import io.grpc.netty.shaded.io.netty.handler.ssl.SslContextBuilder;
|
||||
import io.grpc.netty.shaded.io.netty.handler.ssl.SslProvider;
|
||||
import io.grpc.testing.GrpcCleanupRule;
|
||||
import io.netty.handler.ssl.ApplicationProtocolNames;
|
||||
import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
|
||||
import io.netty.handler.ssl.util.SelfSignedCertificate;
|
||||
import java.io.IOException;
|
||||
import java.security.cert.CertificateException;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import org.apache.rocketmq.proxy.config.ConfigurationManager;
|
||||
import org.apache.rocketmq.proxy.grpc.interceptor.ContextInterceptor;
|
||||
import org.apache.rocketmq.proxy.grpc.interceptor.HeaderInterceptor;
|
||||
import org.apache.rocketmq.test.base.BaseConf;
|
||||
import org.junit.Rule;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
public class GrpcBaseTest extends BaseConf {
|
||||
/**
|
||||
* This rule manages automatic graceful shutdown for the registered servers and channels at the end of test.
|
||||
*/
|
||||
@Rule
|
||||
public final GrpcCleanupRule grpcCleanup = new GrpcCleanupRule();
|
||||
|
||||
private static final int defaultQueueNums = 8;
|
||||
|
||||
protected Channel setUpServer(apache.rocketmq.v1.MessagingServiceGrpc.MessagingServiceImplBase serverImpl,
|
||||
int port, boolean enableInterceptor) throws IOException, CertificateException {
|
||||
SelfSignedCertificate selfSignedCertificate = new SelfSignedCertificate();
|
||||
ServerServiceDefinition serviceDefinition = ServerInterceptors.intercept(serverImpl);
|
||||
if (enableInterceptor) {
|
||||
serviceDefinition = ServerInterceptors.intercept(serverImpl, new ContextInterceptor(), new HeaderInterceptor());
|
||||
}
|
||||
// Create a server, add service, start, and register for automatic graceful shutdown.
|
||||
grpcCleanup.register(NettyServerBuilder.forPort(port)
|
||||
.directExecutor()
|
||||
.addService(serviceDefinition)
|
||||
.useTransportSecurity(selfSignedCertificate.certificate(), selfSignedCertificate.privateKey())
|
||||
.build()
|
||||
.start());
|
||||
// Create a client channel and register for automatic graceful shutdown.
|
||||
return grpcCleanup.register(NettyChannelBuilder.forAddress("127.0.0.1", port)
|
||||
.directExecutor()
|
||||
.sslContext(SslContextBuilder
|
||||
.forClient()
|
||||
.sslProvider(SslProvider.OPENSSL)
|
||||
.trustManager(InsecureTrustManagerFactory.INSTANCE)
|
||||
.applicationProtocolConfig(new ApplicationProtocolConfig(
|
||||
ApplicationProtocolConfig.Protocol.ALPN,
|
||||
ApplicationProtocolConfig.SelectorFailureBehavior.NO_ADVERTISE,
|
||||
ApplicationProtocolConfig.SelectedListenerFailureBehavior.ACCEPT,
|
||||
ApplicationProtocolNames.HTTP_2))
|
||||
.build()
|
||||
)
|
||||
.build());
|
||||
}
|
||||
|
||||
public QueryRouteRequest buildQueryRouteRequest(String topic) {
|
||||
return buildQueryRouteRequest(topic, Endpoints.getDefaultInstance());
|
||||
}
|
||||
|
||||
public QueryRouteRequest buildQueryRouteRequest(String topic, Endpoints endpoints) {
|
||||
return QueryRouteRequest.newBuilder()
|
||||
.setTopic(Resource.newBuilder()
|
||||
.setName(topic)
|
||||
.build())
|
||||
.setEndpoints(endpoints)
|
||||
.build();
|
||||
}
|
||||
|
||||
public SendMessageRequest buildSendMessageRequest(String topic, String messageId) {
|
||||
return SendMessageRequest.newBuilder()
|
||||
.setMessage(Message.newBuilder()
|
||||
.setTopic(Resource.newBuilder()
|
||||
.setName(topic)
|
||||
.build())
|
||||
.setSystemAttribute(SystemAttribute.newBuilder()
|
||||
.setMessageId(messageId)
|
||||
.setPartitionId(0)
|
||||
.build())
|
||||
.setBody(ByteString.copyFromUtf8("123"))
|
||||
.build())
|
||||
.build();
|
||||
}
|
||||
|
||||
public SendMessageRequest buildSendDelayMessageRequest(String topic, String messageId, int delayLevel) {
|
||||
// Message message;
|
||||
// message.getSystemAttribute().getTimedDeliveryCase();
|
||||
return SendMessageRequest.newBuilder()
|
||||
.setMessage(Message.newBuilder()
|
||||
.setTopic(Resource.newBuilder()
|
||||
.setName(topic)
|
||||
.build())
|
||||
.setSystemAttribute(SystemAttribute.newBuilder()
|
||||
.setMessageId(messageId)
|
||||
.setPartitionId(0)
|
||||
.setDelayLevel(delayLevel)
|
||||
.build())
|
||||
.setBody(ByteString.copyFromUtf8("123"))
|
||||
.build())
|
||||
.build();
|
||||
}
|
||||
|
||||
public ReceiveMessageRequest buildReceiveMessageRequest(String group, String topic) {
|
||||
return ReceiveMessageRequest.newBuilder()
|
||||
.setGroup(Resource.newBuilder()
|
||||
.setName(group)
|
||||
.build())
|
||||
.setPartition(Partition.newBuilder()
|
||||
.setTopic(Resource.newBuilder()
|
||||
.setName(topic)
|
||||
.build())
|
||||
.setId(0)
|
||||
.build())
|
||||
.setBatchSize(16)
|
||||
.setInvisibleDuration(Duration.newBuilder()
|
||||
.setSeconds(3)
|
||||
.build())
|
||||
.setInitializationTimestamp(Timestamp.newBuilder()
|
||||
.setSeconds(TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis()))
|
||||
.build())
|
||||
.build();
|
||||
}
|
||||
|
||||
public AckMessageRequest buildAckMessageRequest(String group, String topic, String receiptHandle) {
|
||||
return AckMessageRequest.newBuilder()
|
||||
.setGroup(Resource.newBuilder()
|
||||
.setName(group)
|
||||
.build())
|
||||
.setTopic(Resource.newBuilder()
|
||||
.setName(topic)
|
||||
.build())
|
||||
.setReceiptHandle(receiptHandle)
|
||||
.build();
|
||||
}
|
||||
|
||||
public void assertQueryRoute(QueryRouteResponse response, int brokerSize) {
|
||||
assertThat(response.getCommon().getStatus().getCode()).isEqualTo(Code.OK_VALUE);
|
||||
assertThat(response.getPartitionsList().size()).isEqualTo(brokerSize * defaultQueueNums);
|
||||
assertThat(response.getPartitions(0).getBroker().getEndpoints().getAddresses(0).getPort()).isEqualTo(ConfigurationManager.getProxyConfig().getGrpcServerPort());
|
||||
}
|
||||
|
||||
public void assertSendMessage(SendMessageResponse response, String messageId) {
|
||||
assertThat(response.getCommon()
|
||||
.getStatus()
|
||||
.getCode()).isEqualTo(Code.OK.getNumber());
|
||||
assertThat(response.getMessageId()).isEqualTo(messageId);
|
||||
}
|
||||
|
||||
public void assertReceiveMessage(ReceiveMessageResponse response, String messageId) {
|
||||
assertThat(response.getCommon()
|
||||
.getStatus()
|
||||
.getCode()).isEqualTo(Code.OK.getNumber());
|
||||
assertThat(response.getMessagesCount()).isEqualTo(1);
|
||||
assertThat(response.getMessages(0)
|
||||
.getSystemAttribute()
|
||||
.getMessageId()).isEqualTo(messageId);
|
||||
}
|
||||
|
||||
public void assertAck(AckMessageResponse response) {
|
||||
assertThat(response.getCommon()
|
||||
.getStatus()
|
||||
.getCode()).isEqualTo(Code.OK_VALUE);
|
||||
}
|
||||
}
|
||||
@@ -500,6 +500,7 @@ public class GrpcBaseTest extends BaseConf {
|
||||
|
||||
public QueryRouteRequest buildQueryRouteRequest(String topic) {
|
||||
return QueryRouteRequest.newBuilder()
|
||||
.setEndpoints(buildEndpoints(PORT))
|
||||
.setTopic(Resource.newBuilder()
|
||||
.setName(topic)
|
||||
.build())
|
||||
@@ -508,6 +509,7 @@ public class GrpcBaseTest extends BaseConf {
|
||||
|
||||
public QueryAssignmentRequest buildQueryAssignmentRequest(String topic, String group) {
|
||||
return QueryAssignmentRequest.newBuilder()
|
||||
.setEndpoints(buildEndpoints(PORT))
|
||||
.setTopic(Resource.newBuilder().setName(topic).build())
|
||||
.setGroup(Resource.newBuilder().setName(group).build())
|
||||
.build();
|
||||
@@ -683,15 +685,19 @@ public class GrpcBaseTest extends BaseConf {
|
||||
assertThat(response.getReceiptHandle()).isNotEqualTo(prevHandle);
|
||||
}
|
||||
|
||||
public Endpoints buildEndpoints(int port) {
|
||||
return Endpoints.newBuilder()
|
||||
.setScheme(AddressScheme.IPv4)
|
||||
.addAddresses(Address.newBuilder()
|
||||
.setHost("127.0.0.1")
|
||||
.setPort(port)
|
||||
.build())
|
||||
.build();
|
||||
}
|
||||
|
||||
public Settings buildAccessPointClientSettings(int port) {
|
||||
return Settings.newBuilder()
|
||||
.setAccessPoint(Endpoints.newBuilder()
|
||||
.setScheme(AddressScheme.IPv4)
|
||||
.addAddresses(Address.newBuilder()
|
||||
.setHost("127.0.0.1")
|
||||
.setPort(port)
|
||||
.build())
|
||||
.build())
|
||||
.setAccessPoint(buildEndpoints(port))
|
||||
.build();
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user