@Override
    public CompletableFuture<ProxyRelayResult<ConsumeMessageDirectlyResult>> processConsumeMessageDirectly(
        ProxyContext context, RemotingCommand command,
        ConsumeMessageDirectlyResultRequestHeader header) {
        CompletableFuture<ProxyRelayResult<ConsumeMessageDirectlyResult>> future = new CompletableFuture<>();
        future.thenAccept(proxyOutResult -> {
            RemotingServer remotingServer = this.brokerController.getRemotingServer();
            if (remotingServer instanceof NettyRemotingAbstract) {
                NettyRemotingAbstract nettyRemotingAbstract = (NettyRemotingAbstract) remotingServer;
                RemotingCommand remotingCommand = RemotingCommand.createResponseCommand(null);
                remotingCommand.setOpaque(command.getOpaque());
                remotingCommand.setCode(proxyOutResult.getCode());
                remotingCommand.setRemark(proxyOutResult.getRemark());
                if (proxyOutResult.getCode() == ResponseCode.SUCCESS && proxyOutResult.getResult() != null) {
                    ConsumeMessageDirectlyResult consumeMessageDirectlyResult = proxyOutResult.getResult();
                    remotingCommand.setBody(consumeMessageDirectlyResult.encode());
                }
                SimpleChannel simpleChannel = new SimpleChannel(context.getRemoteAddress(), context.getLocalAddress());
                nettyRemotingAbstract.processResponseCommand(simpleChannel.getChannelHandlerContext(), remotingCommand);
            }
        });
        return future;
    }
@Test
    public void testProcessConsumeMessageDirectly() {
        ConsumeMessageDirectlyResultRequestHeader requestHeader = new ConsumeMessageDirectlyResultRequestHeader();
        String remark = "ok";
        int opaque = 123;
        RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.CONSUME_MESSAGE_DIRECTLY, null);
        remotingCommand.setOpaque(opaque);
        ConsumeMessageDirectlyResult result = new ConsumeMessageDirectlyResult();
        result.setConsumeResult(CMResult.CR_SUCCESS);
        ArgumentCaptor<RemotingCommand> argumentCaptor = ArgumentCaptor.forClass(RemotingCommand.class);
        CompletableFuture<ProxyRelayResult<ConsumeMessageDirectlyResult>> future =
            localProxyRelayService.processConsumeMessageDirectly(ProxyContext.create(), remotingCommand, requestHeader);
        future.complete(new ProxyRelayResult<>(ResponseCode.SUCCESS, remark, result));
        Mockito.verify(nettyRemotingServerMock, Mockito.times(1))
            .processResponseCommand(Mockito.any(SimpleChannelHandlerContext.class), argumentCaptor.capture());
        RemotingCommand remotingCommand1 = argumentCaptor.getValue();
        assertThat(remotingCommand1.getCode()).isEqualTo(ResponseCode.SUCCESS);
        assertThat(remotingCommand1.getRemark()).isEqualTo(remark);
        assertThat(remotingCommand1.getBody()).isEqualTo(result.encode());
    }