private RemotingCommand consumeMessageDirectly(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        final ConsumeMessageDirectlyResultRequestHeader requestHeader =
            (ConsumeMessageDirectlyResultRequestHeader) request
                .decodeCommandCustomHeader(ConsumeMessageDirectlyResultRequestHeader.class);

        final MessageExt msg = MessageDecoder.clientDecode(ByteBuffer.wrap(request.getBody()), true);

        ConsumeMessageDirectlyResult result =
            this.mqClientFactory.consumeMessageDirectly(msg, requestHeader.getConsumerGroup(), requestHeader.getBrokerName());

        if (null != result) {
            response.setCode(ResponseCode.SUCCESS);
            response.setBody(result.encode());
        } else {
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark(String.format("The Consumer Group <%s> not exist in this consumer", requestHeader.getConsumerGroup()));
        }

        return response;
    }
@Test
    public void testConsumeMessageDirectly() throws Exception {
        ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
        RemotingCommand request = mock(RemotingCommand.class);
        when(request.getCode()).thenReturn(RequestCode.CONSUME_MESSAGE_DIRECTLY);
        when(request.getBody()).thenReturn(getMessageResult());
        ConsumeMessageDirectlyResult directlyResult = mock(ConsumeMessageDirectlyResult.class);
        when(mQClientFactory.consumeMessageDirectly(any(MessageExt.class), anyString(), anyString())).thenReturn(directlyResult);
        ConsumeMessageDirectlyResultRequestHeader requestHeader = new ConsumeMessageDirectlyResultRequestHeader();
        requestHeader.setConsumerGroup(defaultGroup);
        requestHeader.setBrokerName(defaultBroker);
        when(request.decodeCommandCustomHeader(ConsumeMessageDirectlyResultRequestHeader.class)).thenReturn(requestHeader);
        RemotingCommand command = processor.processRequest(ctx, request);
        assertNotNull(command);
        assertEquals(ResponseCode.SUCCESS, command.getCode());
    }