opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-01 15:47:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #8901] Add more test coverage for RpcClientImpl (#8902)

Browse Source
This commit is contained in:
yx9o
2024-11-18 10:19:48 +08:00
committed by GitHub
parent 4574647a3a
commit 2e1ca053d2
1 changed files with 239 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
remoting/src/test/java/org/apache/rocketmq/remoting/rpc/RpcClientImplTest.java
+239
View File
@@ -0,0 +1,239 @@
/*
* 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.remoting.rpc;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.InvokeCallback;
import org.apache.rocketmq.remoting.RemotingClient;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
import org.apache.rocketmq.remoting.protocol.header.GetEarliestMsgStoretimeResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.GetMaxOffsetResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.GetMinOffsetResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.PullMessageResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.QueryConsumerOffsetResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.SearchOffsetResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.UpdateConsumerOffsetResponseHeader;
import org.apache.rocketmq.remoting.protocol.statictopic.TopicConfigAndQueueMapping;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import java.util.concurrent.Future;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
public class RpcClientImplTest {
@Mock
private RemotingClient remotingClient;
@Mock
private ClientMetadata clientMetadata;
private RpcClientImpl rpcClient;
private MessageQueue mq;
@Mock
private RpcRequest request;
private final long defaultTimeout = 3000L;
@Before
public void init() throws IllegalAccessException {
rpcClient = new RpcClientImpl(clientMetadata, remotingClient);
String defaultBroker = "brokerName";
mq = new MessageQueue("defaultTopic", defaultBroker, 0);
RpcRequestHeader header = mock(RpcRequestHeader.class);
when(request.getHeader()).thenReturn(header);
when(clientMetadata.getBrokerNameFromMessageQueue(mq)).thenReturn(defaultBroker);
when(clientMetadata.findMasterBrokerAddr(any())).thenReturn("127.0.0.1:10911");
}
@Test
public void testInvoke_PULL_MESSAGE() throws Exception {
when(request.getCode()).thenReturn(RequestCode.PULL_MESSAGE);
doAnswer(invocation -> {
InvokeCallback callback = invocation.getArgument(3);
RemotingCommand response = mock(RemotingCommand.class);
when(response.getBody()).thenReturn("success".getBytes());
PullMessageResponseHeader responseHeader = mock(PullMessageResponseHeader.class);
when(response.decodeCommandCustomHeader(PullMessageResponseHeader.class)).thenReturn(responseHeader);
callback.operationSucceed(response);
return null;
}).when(remotingClient).invokeAsync(
any(),
any(RemotingCommand.class),
anyLong(),
any(InvokeCallback.class));
Future<RpcResponse> future = rpcClient.invoke(mq, request, defaultTimeout);
RpcResponse actual = future.get();
assertEquals(ResponseCode.SUCCESS, actual.getCode());
assertEquals("success", new String((byte[]) actual.getBody()));
}
@Test
public void testInvoke_GET_MIN_OFFSET() throws Exception {
when(request.getCode()).thenReturn(RequestCode.GET_MIN_OFFSET);
RemotingCommand responseCommand = mock(RemotingCommand.class);
when(responseCommand.getBody()).thenReturn("1".getBytes());
GetMinOffsetResponseHeader responseHeader = mock(GetMinOffsetResponseHeader.class);
when(responseCommand.decodeCommandCustomHeader(GetMinOffsetResponseHeader.class)).thenReturn(responseHeader);
when(remotingClient.invokeSync(any(), any(RemotingCommand.class), anyLong())).thenReturn(responseCommand);
Future<RpcResponse> future = rpcClient.invoke(mq, request, defaultTimeout);
RpcResponse actual = future.get();
assertEquals(ResponseCode.SUCCESS, actual.getCode());
assertEquals("1", new String((byte[]) actual.getBody()));
}
@Test
public void testInvoke_GET_MAX_OFFSET() throws Exception {
when(request.getCode()).thenReturn(RequestCode.GET_MAX_OFFSET);
RemotingCommand responseCommand = mock(RemotingCommand.class);
when(responseCommand.getBody()).thenReturn("1000".getBytes());
GetMaxOffsetResponseHeader responseHeader = mock(GetMaxOffsetResponseHeader.class);
when(responseCommand.decodeCommandCustomHeader(GetMaxOffsetResponseHeader.class)).thenReturn(responseHeader);
when(remotingClient.invokeSync(any(), any(RemotingCommand.class), anyLong())).thenReturn(responseCommand);
Future<RpcResponse> future = rpcClient.invoke(mq, request, defaultTimeout);
RpcResponse actual = future.get();
assertEquals(ResponseCode.SUCCESS, actual.getCode());
assertEquals("1000", new String((byte[]) actual.getBody()));
}
@Test
public void testInvoke_SEARCH_OFFSET_BY_TIMESTAMP() throws Exception {
when(request.getCode()).thenReturn(RequestCode.SEARCH_OFFSET_BY_TIMESTAMP);
RemotingCommand responseCommand = mock(RemotingCommand.class);
when(responseCommand.getBody()).thenReturn("1000".getBytes());
SearchOffsetResponseHeader responseHeader = mock(SearchOffsetResponseHeader.class);
when(responseCommand.decodeCommandCustomHeader(SearchOffsetResponseHeader.class)).thenReturn(responseHeader);
when(remotingClient.invokeSync(any(), any(RemotingCommand.class), anyLong())).thenReturn(responseCommand);
Future<RpcResponse> future = rpcClient.invoke(mq, request, defaultTimeout);
RpcResponse actual = future.get();
assertEquals(ResponseCode.SUCCESS, actual.getCode());
assertEquals("1000", new String((byte[]) actual.getBody()));
}
@Test
public void testInvoke_GET_EARLIEST_MSG_STORETIME() throws Exception {
when(request.getCode()).thenReturn(RequestCode.GET_EARLIEST_MSG_STORETIME);
RemotingCommand responseCommand = mock(RemotingCommand.class);
when(responseCommand.getBody()).thenReturn("10000".getBytes());
GetEarliestMsgStoretimeResponseHeader responseHeader = mock(GetEarliestMsgStoretimeResponseHeader.class);
when(responseCommand.decodeCommandCustomHeader(GetEarliestMsgStoretimeResponseHeader.class)).thenReturn(responseHeader);
when(remotingClient.invokeSync(any(), any(RemotingCommand.class), anyLong())).thenReturn(responseCommand);
Future<RpcResponse> future = rpcClient.invoke(mq, request, defaultTimeout);
RpcResponse actual = future.get();
assertEquals(ResponseCode.SUCCESS, actual.getCode());
assertEquals("10000", new String((byte[]) actual.getBody()));
}
@Test
public void testInvoke_QUERY_CONSUMER_OFFSET() throws Exception {
when(request.getCode()).thenReturn(RequestCode.QUERY_CONSUMER_OFFSET);
RemotingCommand responseCommand = mock(RemotingCommand.class);
when(responseCommand.getBody()).thenReturn("1000".getBytes());
QueryConsumerOffsetResponseHeader responseHeader = mock(QueryConsumerOffsetResponseHeader.class);
when(responseCommand.decodeCommandCustomHeader(QueryConsumerOffsetResponseHeader.class)).thenReturn(responseHeader);
when(remotingClient.invokeSync(any(), any(RemotingCommand.class), anyLong())).thenReturn(responseCommand);
Future<RpcResponse> future = rpcClient.invoke(mq, request, defaultTimeout);
RpcResponse actual = future.get();
assertEquals(ResponseCode.SUCCESS, actual.getCode());
assertEquals("1000", new String((byte[]) actual.getBody()));
}
@Test
public void testInvoke_UPDATE_CONSUMER_OFFSET() throws Exception {
when(request.getCode()).thenReturn(RequestCode.UPDATE_CONSUMER_OFFSET);
RemotingCommand responseCommand = mock(RemotingCommand.class);
when(responseCommand.getBody()).thenReturn("success".getBytes());
UpdateConsumerOffsetResponseHeader responseHeader = mock(UpdateConsumerOffsetResponseHeader.class);
when(responseCommand.decodeCommandCustomHeader(UpdateConsumerOffsetResponseHeader.class)).thenReturn(responseHeader);
when(remotingClient.invokeSync(any(), any(RemotingCommand.class), anyLong())).thenReturn(responseCommand);
Future<RpcResponse> future = rpcClient.invoke(mq, request, defaultTimeout);
RpcResponse actual = future.get();
assertEquals(ResponseCode.SUCCESS, actual.getCode());
assertEquals("success", new String((byte[]) actual.getBody()));
}
@Test
public void testInvoke_GET_TOPIC_STATS_INFO() throws Exception {
when(request.getCode()).thenReturn(RequestCode.GET_TOPIC_STATS_INFO);
RemotingCommand responseCommand = mock(RemotingCommand.class);
TopicStatsTable topicStatsTable = new TopicStatsTable();
when(responseCommand.getBody()).thenReturn(topicStatsTable.encode());
when(remotingClient.invokeSync(any(), any(RemotingCommand.class), anyLong())).thenReturn(responseCommand);
Future<RpcResponse> future = rpcClient.invoke(mq, request, defaultTimeout);
RpcResponse actual = future.get();
assertEquals(ResponseCode.SUCCESS, actual.getCode());
assertTrue(actual.getBody() instanceof TopicStatsTable);
}
@Test
public void testInvoke_GET_TOPIC_CONFIG() throws Exception {
when(request.getCode()).thenReturn(RequestCode.GET_TOPIC_CONFIG);
RemotingCommand responseCommand = mock(RemotingCommand.class);
TopicConfigAndQueueMapping topicConfigAndQueueMapping = new TopicConfigAndQueueMapping();
when(responseCommand.getBody()).thenReturn(RemotingSerializable.encode(topicConfigAndQueueMapping));
when(remotingClient.invokeSync(any(), any(RemotingCommand.class), anyLong())).thenReturn(responseCommand);
Future<RpcResponse> future = rpcClient.invoke(mq, request, defaultTimeout);
RpcResponse actual = future.get();
assertEquals(ResponseCode.SUCCESS, actual.getCode());
assertTrue(actual.getBody() instanceof TopicConfigAndQueueMapping);
}
}