@Override
    public ChannelFuture writeAndFlush(Object msg) {
        CompletableFuture<Void> processFuture = new CompletableFuture<>();

        try {
            if (msg instanceof RemotingCommand) {
                ProxyContext context = ProxyContext.createForInner(this.getClass())
                    .setRemoteAddress(remoteAddress)
                    .setLocalAddress(localAddress);
                RemotingCommand command = (RemotingCommand) msg;
                if (command.getExtFields() == null) {
                    command.setExtFields(new HashMap<>());
                }
                switch (command.getCode()) {
                    case RequestCode.CHECK_TRANSACTION_STATE: {
                        CheckTransactionStateRequestHeader header = (CheckTransactionStateRequestHeader) command.readCustomHeader();
                        MessageExt messageExt = MessageDecoder.decode(ByteBuffer.wrap(command.getBody()), true, false, false);
                        RelayData<TransactionData, Void> relayData = this.proxyRelayService.processCheckTransactionState(context, command, header, messageExt);
                        processFuture = this.processCheckTransaction(header, messageExt, relayData.getProcessResult(), relayData.getRelayFuture());
                        break;
                    }
                    case RequestCode.GET_CONSUMER_RUNNING_INFO: {
                        GetConsumerRunningInfoRequestHeader header = (GetConsumerRunningInfoRequestHeader) command.readCustomHeader();
                        CompletableFuture<ProxyRelayResult<ConsumerRunningInfo>> relayFuture = this.proxyRelayService.processGetConsumerRunningInfo(context, command, header);
                        processFuture = this.processGetConsumerRunningInfo(command, header, relayFuture);
                        break;
                    }
                    case RequestCode.CONSUME_MESSAGE_DIRECTLY: {
                        ConsumeMessageDirectlyResultRequestHeader header = (ConsumeMessageDirectlyResultRequestHeader) command.readCustomHeader();
                        MessageExt messageExt = MessageDecoder.decode(ByteBuffer.wrap(command.getBody()), true, false, false);
                        processFuture = this.processConsumeMessageDirectly(command, header, messageExt,
                            this.proxyRelayService.processConsumeMessageDirectly(context, command, header));
                        break;
                    }
                    default:
                        break;
                }
            } else {
                processFuture = processOtherMessage(msg);
            }
        } catch (Throwable t) {
            log.error("process failed. msg:{}", msg, t);
            processFuture.completeExceptionally(t);
        }

        DefaultChannelPromise promise = new DefaultChannelPromise(this, GlobalEventExecutor.INSTANCE);
        processFuture.thenAccept(ignore -> promise.setSuccess())
            .exceptionally(t -> {
                promise.setFailure(t);
                return null;
            });
        return promise;
    }
@Test
    public void testWriteAndFlush() throws Exception {
        when(this.proxyRelayService.processCheckTransactionState(any(), any(), any(), any()))
            .thenReturn(new RelayData<>(mock(TransactionData.class), new CompletableFuture<>()));

        ArgumentCaptor<ConsumeMessageDirectlyResultRequestHeader> consumeMessageDirectlyArgumentCaptor =
            ArgumentCaptor.forClass(ConsumeMessageDirectlyResultRequestHeader.class);
        when(this.proxyRelayService.processConsumeMessageDirectly(any(), any(), consumeMessageDirectlyArgumentCaptor.capture()))
            .thenReturn(new CompletableFuture<>());

        ArgumentCaptor<GetConsumerRunningInfoRequestHeader> getConsumerRunningInfoArgumentCaptor =
            ArgumentCaptor.forClass(GetConsumerRunningInfoRequestHeader.class);
        when(this.proxyRelayService.processGetConsumerRunningInfo(any(), any(), getConsumerRunningInfoArgumentCaptor.capture()))
            .thenReturn(new CompletableFuture<>());

        CheckTransactionStateRequestHeader checkTransactionStateRequestHeader = new CheckTransactionStateRequestHeader();
        checkTransactionStateRequestHeader.setTransactionId(MessageClientIDSetter.createUniqID());
        RemotingCommand checkTransactionRequest = RemotingCommand.createRequestCommand(RequestCode.CHECK_TRANSACTION_STATE, checkTransactionStateRequestHeader);
        MessageExt transactionMessageExt = new MessageExt();
        transactionMessageExt.setTopic("topic");
        transactionMessageExt.setTags("tags");
        transactionMessageExt.setBornHost(NetworkUtil.string2SocketAddress("127.0.0.2:8888"));
        transactionMessageExt.setStoreHost(NetworkUtil.string2SocketAddress("127.0.0.1:10911"));
        transactionMessageExt.setBody(UUID.randomUUID().toString().getBytes(StandardCharsets.UTF_8));
        transactionMessageExt.setMsgId(MessageClientIDSetter.createUniqID());
        checkTransactionRequest.setBody(MessageDecoder.encode(transactionMessageExt, false));

        GetConsumerRunningInfoRequestHeader consumerRunningInfoRequestHeader = new GetConsumerRunningInfoRequestHeader();
        consumerRunningInfoRequestHeader.setConsumerGroup("group");
        consumerRunningInfoRequestHeader.setClientId("clientId");
        RemotingCommand consumerRunningInfoRequest = RemotingCommand.createRequestCommand(RequestCode.GET_CONSUMER_RUNNING_INFO, consumerRunningInfoRequestHeader);

        ConsumeMessageDirectlyResultRequestHeader consumeMessageDirectlyResultRequestHeader = new ConsumeMessageDirectlyResultRequestHeader();
        consumeMessageDirectlyResultRequestHeader.setConsumerGroup("group");
        consumeMessageDirectlyResultRequestHeader.setClientId("clientId");
        MessageExt consumeMessageDirectlyMessageExt = new MessageExt();
        consumeMessageDirectlyMessageExt.setTopic("topic");
        consumeMessageDirectlyMessageExt.setTags("tags");
        consumeMessageDirectlyMessageExt.setBornHost(NetworkUtil.string2SocketAddress("127.0.0.2:8888"));
        consumeMessageDirectlyMessageExt.setStoreHost(NetworkUtil.string2SocketAddress("127.0.0.1:10911"));
        consumeMessageDirectlyMessageExt.setBody(UUID.randomUUID().toString().getBytes(StandardCharsets.UTF_8));
        consumeMessageDirectlyMessageExt.setMsgId(MessageClientIDSetter.createUniqID());
        RemotingCommand consumeMessageDirectlyResult = RemotingCommand.createRequestCommand(RequestCode.CONSUME_MESSAGE_DIRECTLY, consumeMessageDirectlyResultRequestHeader);
        consumeMessageDirectlyResult.setBody(MessageDecoder.encode(consumeMessageDirectlyMessageExt, false));

        MockProxyChannel channel = new MockProxyChannel(this.proxyRelayService, null, "127.0.0.2:8888", "127.0.0.1:10911") {
            @Override
            protected CompletableFuture<Void> processOtherMessage(Object msg) {
                return CompletableFuture.completedFuture(null);
            }

            @Override
            protected CompletableFuture<Void> processCheckTransaction(CheckTransactionStateRequestHeader header,
                MessageExt messageExt, TransactionData transactionData, CompletableFuture<ProxyRelayResult<Void>> responseFuture) {
                assertEquals(checkTransactionStateRequestHeader, header);
                assertArrayEquals(transactionMessageExt.getBody(), messageExt.getBody());
                return CompletableFuture.completedFuture(null);
            }

            @Override
            protected CompletableFuture<Void> processGetConsumerRunningInfo(RemotingCommand command,
                GetConsumerRunningInfoRequestHeader header,
                CompletableFuture<ProxyRelayResult<ConsumerRunningInfo>> responseFuture) {
                assertEquals(consumerRunningInfoRequestHeader, getConsumerRunningInfoArgumentCaptor.getValue());
                assertEquals(consumerRunningInfoRequestHeader, header);
                return CompletableFuture.completedFuture(null);
            }

            @Override
            protected CompletableFuture<Void> processConsumeMessageDirectly(RemotingCommand command,
                ConsumeMessageDirectlyResultRequestHeader header, MessageExt messageExt,
                CompletableFuture<ProxyRelayResult<ConsumeMessageDirectlyResult>> responseFuture) {
                assertEquals(consumeMessageDirectlyResultRequestHeader, consumeMessageDirectlyArgumentCaptor.getValue());
                assertEquals(consumeMessageDirectlyResultRequestHeader, header);
                assertArrayEquals(consumeMessageDirectlyMessageExt.getBody(), messageExt.getBody());
                return CompletableFuture.completedFuture(null);
            }
        };

        assertTrue(channel.writeAndFlush(checkTransactionRequest).isSuccess());
        assertTrue(channel.writeAndFlush(consumerRunningInfoRequest).isSuccess());
        assertTrue(channel.writeAndFlush(consumeMessageDirectlyResult).isSuccess());
    }