@Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        SendMessageContext sendMessageContext;
        switch (request.getCode()) {
            case RequestCode.CONSUMER_SEND_MSG_BACK:
                return this.consumerSendMsgBack(ctx, request);
            default:
                SendMessageRequestHeader requestHeader = parseRequestHeader(request);
                if (requestHeader == null) {
                    return null;
                }
                TopicQueueMappingContext mappingContext = this.brokerController.getTopicQueueMappingManager().buildTopicQueueMappingContext(requestHeader, true);
                RemotingCommand rewriteResult = this.brokerController.getTopicQueueMappingManager().rewriteRequestForStaticTopic(requestHeader, mappingContext);
                if (rewriteResult != null) {
                    return rewriteResult;
                }
                sendMessageContext = buildMsgContext(ctx, requestHeader, request);
                try {
                    this.executeSendMessageHookBefore(sendMessageContext);
                } catch (AbortProcessException e) {
                    final RemotingCommand errorResponse = RemotingCommand.createResponseCommand(e.getResponseCode(), e.getErrorMessage());
                    errorResponse.setOpaque(request.getOpaque());
                    return errorResponse;
                }

                RemotingCommand response;
                clearReservedProperties(requestHeader);

                if (requestHeader.isBatch()) {
                    response = this.sendBatchMessage(ctx, request, sendMessageContext, requestHeader, mappingContext,
                        (ctx1, response1) -> executeSendMessageHookAfter(response1, ctx1));
                } else {
                    response = this.sendMessage(ctx, request, sendMessageContext, requestHeader, mappingContext,
                        (ctx12, response12) -> executeSendMessageHookAfter(response12, ctx12));
                }

                return response;
        }
    }
@Test
    public void testProcessRequest_Transaction() throws RemotingCommandException {
        brokerController.setTransactionalMessageService(transactionMsgService);
        when(brokerController.getTransactionalMessageService().asyncPrepareMessage(any(MessageExtBrokerInner.class)))
            .thenReturn(CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.PUT_OK, new AppendMessageResult(AppendMessageStatus.PUT_OK))));
        RemotingCommand request = createSendTransactionMsgCommand(RequestCode.SEND_MESSAGE);
        final RemotingCommand[] response = new RemotingCommand[1];
        doAnswer(invocation -> {
            response[0] = invocation.getArgument(0);
            return null;
        }).when(channel).writeAndFlush(any(Object.class));
        await().atMost(Duration.ofSeconds(10)).until(() -> {
            RemotingCommand responseToReturn = sendMessageProcessor.processRequest(handlerContext, request);
            if (responseToReturn != null) {
                assertThat(response[0]).isNull();
                response[0] = responseToReturn;
            }

            if (response[0] == null) {
                return false;
            }
            assertThat(response[0].getCode()).isEqualTo(ResponseCode.SUCCESS);
            assertThat(response[0].getOpaque()).isEqualTo(request.getOpaque());
            return true;
        });
    }