### Question:
PullMessageProcessor implements NettyRequestProcessor { @Override public RemotingCommand processRequest(final ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException { return this.processRequest(ctx.channel(), request, true); } PullMessageProcessor(final BrokerController brokerController); @Override RemotingCommand processRequest(final ChannelHandlerContext ctx,
        RemotingCommand request); @Override boolean rejectRequest(); boolean hasConsumeMessageHook(); void executeConsumeMessageHookBefore(final ConsumeMessageContext context); void executeRequestWhenWakeup(final Channel channel,
        final RemotingCommand request); void registerConsumeMessageHook(List<ConsumeMessageHook> sendMessageHookList);  }### Answer:
@Test public void testProcessRequest_TopicNotExist() throws RemotingCommandException { brokerController.getTopicConfigManager().getTopicConfigTable().remove(topic); final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE); RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request); assertThat(response).isNotNull(); assertThat(response.getCode()).isEqualTo(ResponseCode.TOPIC_NOT_EXIST); assertThat(response.getRemark()).contains("topic[" + topic + "] not exist"); }
@Test public void testProcessRequest_SubNotExist() throws RemotingCommandException { brokerController.getConsumerManager().unregisterConsumer(group, clientChannelInfo, false); final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE); RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request); assertThat(response).isNotNull(); assertThat(response.getCode()).isEqualTo(ResponseCode.SUBSCRIPTION_NOT_EXIST); assertThat(response.getRemark()).contains("consumer's group info not exist"); }
@Test public void testProcessRequest_SubNotLatest() throws RemotingCommandException { final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE); request.addExtField("subVersion", String.valueOf(101)); RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request); assertThat(response).isNotNull(); assertThat(response.getCode()).isEqualTo(ResponseCode.SUBSCRIPTION_NOT_LATEST); assertThat(response.getRemark()).contains("subscription not latest"); }
@Test public void testProcessRequest_Found() throws RemotingCommandException { GetMessageResult getMessageResult = createGetMessageResult(); when(messageStore.getMessage(anyString(), anyString(), anyInt(), anyLong(), anyInt(), any(ExpressionMessageFilter.class))).thenReturn(getMessageResult); final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE); RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request); assertThat(response).isNotNull(); assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS); }
@Test public void testProcessRequest_MsgWasRemoving() throws RemotingCommandException { GetMessageResult getMessageResult = createGetMessageResult(); getMessageResult.setStatus(GetMessageStatus.MESSAGE_WAS_REMOVING); when(messageStore.getMessage(anyString(), anyString(), anyInt(), anyLong(), anyInt(), any(ExpressionMessageFilter.class))).thenReturn(getMessageResult); final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE); RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request); assertThat(response).isNotNull(); assertThat(response.getCode()).isEqualTo(ResponseCode.PULL_RETRY_IMMEDIATELY); }
@Test public void testProcessRequest_NoMsgInQueue() throws RemotingCommandException { GetMessageResult getMessageResult = createGetMessageResult(); getMessageResult.setStatus(GetMessageStatus.NO_MESSAGE_IN_QUEUE); when(messageStore.getMessage(anyString(), anyString(), anyInt(), anyLong(), anyInt(), any(ExpressionMessageFilter.class))).thenReturn(getMessageResult); final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE); RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request); assertThat(response).isNotNull(); assertThat(response.getCode()).isEqualTo(ResponseCode.PULL_OFFSET_MOVED); }