@Override
    public List<DefaultAuthorizationContext> build(Metadata metadata, GeneratedMessageV3 message) {
        List<DefaultAuthorizationContext> result = null;
        if (message instanceof SendMessageRequest) {
            SendMessageRequest request = (SendMessageRequest) message;
            if (request.getMessagesCount() <= 0) {
                throw new AuthorizationException("message is null.");
            }
            result = newPubContext(metadata, request.getMessages(0).getTopic());
        }
        if (message instanceof EndTransactionRequest) {
            EndTransactionRequest request = (EndTransactionRequest) message;
            result = newPubContext(metadata, request.getTopic());
        }
        if (message instanceof HeartbeatRequest) {
            HeartbeatRequest request = (HeartbeatRequest) message;
            if (!isConsumerClientType(request.getClientType())) {
                return null;
            }
            result = newGroupSubContexts(metadata, request.getGroup());
        }
        if (message instanceof ReceiveMessageRequest) {
            ReceiveMessageRequest request = (ReceiveMessageRequest) message;
            if (!request.hasMessageQueue()) {
                throw new AuthorizationException("messageQueue is null.");
            }
            result = newSubContexts(metadata, request.getGroup(), request.getMessageQueue().getTopic());
        }
        if (message instanceof AckMessageRequest) {
            AckMessageRequest request = (AckMessageRequest) message;
            result = newSubContexts(metadata, request.getGroup(), request.getTopic());
        }
        if (message instanceof ForwardMessageToDeadLetterQueueRequest) {
            ForwardMessageToDeadLetterQueueRequest request = (ForwardMessageToDeadLetterQueueRequest) message;
            result = newSubContexts(metadata, request.getGroup(), request.getTopic());
        }
        if (message instanceof NotifyClientTerminationRequest) {
            NotifyClientTerminationRequest request = (NotifyClientTerminationRequest) message;
            if (StringUtils.isNotBlank(request.getGroup().getName())) {
                result = newGroupSubContexts(metadata, request.getGroup());
            }
        }
        if (message instanceof ChangeInvisibleDurationRequest) {
            ChangeInvisibleDurationRequest request = (ChangeInvisibleDurationRequest) message;
            result = newGroupSubContexts(metadata, request.getGroup());
        }
        if (message instanceof QueryRouteRequest) {
            QueryRouteRequest request = (QueryRouteRequest) message;
            result = newContext(metadata, request);
        }
        if (message instanceof QueryAssignmentRequest) {
            QueryAssignmentRequest request = (QueryAssignmentRequest) message;
            result = newSubContexts(metadata, request.getGroup(), request.getTopic());
        }
        if (message instanceof TelemetryCommand) {
            TelemetryCommand request = (TelemetryCommand) message;
            result = newContext(metadata, request);
        }
        if (CollectionUtils.isNotEmpty(result)) {
            result.forEach(context -> {
                context.setChannelId(metadata.get(GrpcConstants.CHANNEL_ID));
                context.setRpcCode(message.getDescriptorForType().getFullName());
            });
        }
        return result;
    }
@Test
    public void buildRemoting() {
        when(channel.id()).thenReturn(mockChannelId("channel-id"));
        when(channel.hasAttr(eq(AttributeKeys.PROXY_PROTOCOL_ADDR))).thenReturn(true);
        when(channel.attr(eq(AttributeKeys.PROXY_PROTOCOL_ADDR))).thenReturn(mockAttribute("192.168.0.1"));
        when(channel.hasAttr(eq(AttributeKeys.PROXY_PROTOCOL_PORT))).thenReturn(true);
        when(channel.attr(eq(AttributeKeys.PROXY_PROTOCOL_PORT))).thenReturn(mockAttribute("1234"));
        when(channelHandlerContext.channel()).thenReturn(channel);

        SendMessageRequestHeader sendMessageRequestHeader = new SendMessageRequestHeader();
        sendMessageRequestHeader.setTopic("topic");
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, sendMessageRequestHeader);
        request.setVersion(441);
        request.addExtField("AccessKey", "rocketmq");
        request.makeCustomHeaderToNet();
        List<DefaultAuthorizationContext> result = builder.build(channelHandlerContext, request);
        Assert.assertEquals(1, result.size());
        Assert.assertEquals("User:rocketmq", result.get(0).getSubject().getSubjectKey());
        Assert.assertEquals("Topic:topic", result.get(0).getResource().getResourceKey());
        Assert.assertTrue(result.get(0).getActions().containsAll(Arrays.asList(Action.PUB)));
        Assert.assertEquals("192.168.0.1", result.get(0).getSourceIp());
        Assert.assertEquals("channel-id", result.get(0).getChannelId());
        Assert.assertEquals(RequestCode.SEND_MESSAGE + "", result.get(0).getRpcCode());

        sendMessageRequestHeader = new SendMessageRequestHeader();
        sendMessageRequestHeader.setTopic("%RETRY%group");
        request = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, sendMessageRequestHeader);
        request.setVersion(441);
        request.addExtField("AccessKey", "rocketmq");
        request.makeCustomHeaderToNet();
        result = builder.build(channelHandlerContext, request);
        Assert.assertEquals(1, result.size());
        Assert.assertEquals("User:rocketmq", result.get(0).getSubject().getSubjectKey());
        Assert.assertEquals("Group:group", result.get(0).getResource().getResourceKey());
        Assert.assertTrue(result.get(0).getActions().containsAll(Arrays.asList(Action.SUB)));

        SendMessageRequestHeaderV2 sendMessageRequestHeaderV2 = new SendMessageRequestHeaderV2();
        sendMessageRequestHeaderV2.setTopic("topic");
        request = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE_V2, sendMessageRequestHeaderV2);
        request.setVersion(441);
        request.addExtField("AccessKey", "rocketmq");
        request.makeCustomHeaderToNet();
        result = builder.build(channelHandlerContext, request);
        Assert.assertEquals(1, result.size());
        Assert.assertEquals("User:rocketmq", result.get(0).getSubject().getSubjectKey());
        Assert.assertEquals("Topic:topic", result.get(0).getResource().getResourceKey());
        Assert.assertTrue(result.get(0).getActions().containsAll(Arrays.asList(Action.PUB)));

        sendMessageRequestHeaderV2 = new SendMessageRequestHeaderV2();
        sendMessageRequestHeaderV2.setTopic("%RETRY%group");
        request = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE_V2, sendMessageRequestHeaderV2);
        request.setVersion(441);
        request.addExtField("AccessKey", "rocketmq");
        request.makeCustomHeaderToNet();
        result = builder.build(channelHandlerContext, request);
        Assert.assertEquals(1, result.size());
        Assert.assertEquals("User:rocketmq", result.get(0).getSubject().getSubjectKey());
        Assert.assertEquals("Group:group", result.get(0).getResource().getResourceKey());
        Assert.assertTrue(result.get(0).getActions().containsAll(Arrays.asList(Action.SUB)));

        EndTransactionRequestHeader endTransactionRequestHeader = new EndTransactionRequestHeader();
        endTransactionRequestHeader.setTopic("topic");
        request = RemotingCommand.createRequestCommand(RequestCode.END_TRANSACTION, endTransactionRequestHeader);
        request.setVersion(441);
        request.addExtField("AccessKey", "rocketmq");
        request.makeCustomHeaderToNet();
        result = builder.build(channelHandlerContext, request);
        Assert.assertEquals(1, result.size());
        Assert.assertEquals("User:rocketmq", result.get(0).getSubject().getSubjectKey());
        Assert.assertEquals("Topic:topic", result.get(0).getResource().getResourceKey());
        Assert.assertTrue(result.get(0).getActions().containsAll(Arrays.asList(Action.PUB)));

        endTransactionRequestHeader = new EndTransactionRequestHeader();
        request = RemotingCommand.createRequestCommand(RequestCode.END_TRANSACTION, endTransactionRequestHeader);
        request.setVersion(441);
        request.addExtField("AccessKey", "rocketmq");
        request.makeCustomHeaderToNet();
        result = builder.build(channelHandlerContext, request);
        Assert.assertEquals(0, result.size());

        ConsumerSendMsgBackRequestHeader consumerSendMsgBackRequestHeader = new ConsumerSendMsgBackRequestHeader();
        consumerSendMsgBackRequestHeader.setGroup("group");
        request = RemotingCommand.createRequestCommand(RequestCode.CONSUMER_SEND_MSG_BACK, consumerSendMsgBackRequestHeader);
        request.setVersion(441);
        request.addExtField("AccessKey", "rocketmq");
        request.makeCustomHeaderToNet();
        result = builder.build(channelHandlerContext, request);
        Assert.assertEquals(1, result.size());
        Assert.assertEquals("User:rocketmq", result.get(0).getSubject().getSubjectKey());
        Assert.assertEquals("Group:group", result.get(0).getResource().getResourceKey());
        Assert.assertTrue(result.get(0).getActions().containsAll(Arrays.asList(Action.SUB)));

        PullMessageRequestHeader pullMessageRequestHeader = new PullMessageRequestHeader();
        pullMessageRequestHeader.setTopic("topic");
        pullMessageRequestHeader.setConsumerGroup("group");
        request = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, pullMessageRequestHeader);
        request.setVersion(441);
        request.addExtField("AccessKey", "rocketmq");
        request.makeCustomHeaderToNet();
        result = builder.build(channelHandlerContext, request);
        Assert.assertEquals(2, result.size());
        Assert.assertEquals("User:rocketmq", getContext(result, ResourceType.GROUP).getSubject().getSubjectKey());
        Assert.assertEquals("Group:group", getContext(result, ResourceType.GROUP).getResource().getResourceKey());
        Assert.assertTrue(getContext(result, ResourceType.GROUP).getActions().containsAll(Arrays.asList(Action.SUB)));
        Assert.assertEquals("User:rocketmq", getContext(result, ResourceType.TOPIC).getSubject().getSubjectKey());
        Assert.assertEquals("Topic:topic", getContext(result, ResourceType.TOPIC).getResource().getResourceKey());
        Assert.assertTrue(getContext(result, ResourceType.TOPIC).getActions().containsAll(Arrays.asList(Action.SUB)));

        QueryMessageRequestHeader queryMessageRequestHeader = new QueryMessageRequestHeader();
        queryMessageRequestHeader.setTopic("topic");
        request = RemotingCommand.createRequestCommand(RequestCode.QUERY_MESSAGE, queryMessageRequestHeader);
        request.setVersion(441);
        request.addExtField("AccessKey", "rocketmq");
        request.makeCustomHeaderToNet();
        result = builder.build(channelHandlerContext, request);
        Assert.assertEquals(1, result.size());
        Assert.assertEquals("User:rocketmq", result.get(0).getSubject().getSubjectKey());
        Assert.assertEquals("Topic:topic", result.get(0).getResource().getResourceKey());
        Assert.assertTrue(result.get(0).getActions().containsAll(Arrays.asList(Action.SUB, Action.GET)));

        HeartbeatRequestHeader heartbeatRequestHeader = new HeartbeatRequestHeader();
        request = RemotingCommand.createRequestCommand(RequestCode.HEART_BEAT, heartbeatRequestHeader);
        HeartbeatData heartbeatData = new HeartbeatData();
        ConsumerData consumerData = new ConsumerData();
        consumerData.setGroupName("group");
        SubscriptionData subscriptionData = new SubscriptionData();
        subscriptionData.setTopic("topic");
        consumerData.setSubscriptionDataSet(Sets.newHashSet(subscriptionData));
        heartbeatData.setConsumerDataSet(Sets.newHashSet(consumerData));
        request.setBody(JSON.toJSONBytes(heartbeatData));
        request.setVersion(441);
        request.addExtField("AccessKey", "rocketmq");
        request.makeCustomHeaderToNet();
        result = builder.build(channelHandlerContext, request);
        Assert.assertEquals(2, result.size());
        Assert.assertEquals("User:rocketmq", getContext(result, ResourceType.GROUP).getSubject().getSubjectKey());
        Assert.assertEquals("Group:group", getContext(result, ResourceType.GROUP).getResource().getResourceKey());
        Assert.assertTrue(getContext(result, ResourceType.GROUP).getActions().containsAll(Arrays.asList(Action.SUB)));
        Assert.assertEquals("User:rocketmq", getContext(result, ResourceType.TOPIC).getSubject().getSubjectKey());
        Assert.assertEquals("Topic:topic", getContext(result, ResourceType.TOPIC).getResource().getResourceKey());
        Assert.assertTrue(getContext(result, ResourceType.TOPIC).getActions().containsAll(Arrays.asList(Action.SUB)));

        UnregisterClientRequestHeader unregisterClientRequestHeader = new UnregisterClientRequestHeader();
        unregisterClientRequestHeader.setConsumerGroup("group");
        request = RemotingCommand.createRequestCommand(RequestCode.UNREGISTER_CLIENT, unregisterClientRequestHeader);
        request.setVersion(441);
        request.addExtField("AccessKey", "rocketmq");
        request.makeCustomHeaderToNet();
        result = builder.build(channelHandlerContext, request);
        Assert.assertEquals(1, result.size());
        Assert.assertEquals("User:rocketmq", result.get(0).getSubject().getSubjectKey());
        Assert.assertEquals("Group:group", result.get(0).getResource().getResourceKey());
        Assert.assertTrue(result.get(0).getActions().containsAll(Arrays.asList(Action.SUB)));

        GetConsumerListByGroupRequestHeader getConsumerListByGroupRequestHeader = new GetConsumerListByGroupRequestHeader();
        getConsumerListByGroupRequestHeader.setConsumerGroup("group");
        request = RemotingCommand.createRequestCommand(RequestCode.GET_CONSUMER_LIST_BY_GROUP, getConsumerListByGroupRequestHeader);
        request.setVersion(441);
        request.addExtField("AccessKey", "rocketmq");
        request.makeCustomHeaderToNet();
        result = builder.build(channelHandlerContext, request);
        Assert.assertEquals(1, result.size());
        Assert.assertEquals("User:rocketmq", result.get(0).getSubject().getSubjectKey());
        Assert.assertEquals("Group:group", result.get(0).getResource().getResourceKey());
        Assert.assertTrue(result.get(0).getActions().containsAll(Arrays.asList(Action.SUB, Action.GET)));

        QueryConsumerOffsetRequestHeader queryConsumerOffsetRequestHeader = new QueryConsumerOffsetRequestHeader();
        queryConsumerOffsetRequestHeader.setTopic("topic");
        queryConsumerOffsetRequestHeader.setConsumerGroup("group");
        request = RemotingCommand.createRequestCommand(RequestCode.QUERY_CONSUMER_OFFSET, queryConsumerOffsetRequestHeader);
        request.setVersion(441);
        request.addExtField("AccessKey", "rocketmq");
        request.makeCustomHeaderToNet();
        result = builder.build(channelHandlerContext, request);
        Assert.assertEquals(2, result.size());
        Assert.assertEquals("User:rocketmq", getContext(result, ResourceType.GROUP).getSubject().getSubjectKey());
        Assert.assertEquals("Group:group", getContext(result, ResourceType.GROUP).getResource().getResourceKey());
        Assert.assertTrue(getContext(result, ResourceType.GROUP).getActions().containsAll(Arrays.asList(Action.SUB)));
        Assert.assertEquals("User:rocketmq", getContext(result, ResourceType.TOPIC).getSubject().getSubjectKey());
        Assert.assertEquals("Topic:topic", getContext(result, ResourceType.TOPIC).getResource().getResourceKey());
        Assert.assertTrue(getContext(result, ResourceType.TOPIC).getActions().containsAll(Arrays.asList(Action.SUB)));

        UpdateConsumerOffsetRequestHeader updateConsumerOffsetRequestHeader = new UpdateConsumerOffsetRequestHeader();
        updateConsumerOffsetRequestHeader.setTopic("topic");
        updateConsumerOffsetRequestHeader.setConsumerGroup("group");
        request = RemotingCommand.createRequestCommand(RequestCode.UPDATE_CONSUMER_OFFSET, updateConsumerOffsetRequestHeader);
        request.setVersion(441);
        request.addExtField("AccessKey", "rocketmq");
        request.makeCustomHeaderToNet();
        result = builder.build(channelHandlerContext, request);
        Assert.assertEquals(2, result.size());
        Assert.assertEquals("User:rocketmq", getContext(result, ResourceType.GROUP).getSubject().getSubjectKey());
        Assert.assertEquals("Group:group", getContext(result, ResourceType.GROUP).getResource().getResourceKey());
        Assert.assertTrue(getContext(result, ResourceType.GROUP).getActions().containsAll(Arrays.asList(Action.SUB, Action.UPDATE)));
        Assert.assertEquals("User:rocketmq", getContext(result, ResourceType.TOPIC).getSubject().getSubjectKey());
        Assert.assertEquals("Topic:topic", getContext(result, ResourceType.TOPIC).getResource().getResourceKey());
        Assert.assertTrue(getContext(result, ResourceType.TOPIC).getActions().containsAll(Arrays.asList(Action.SUB, Action.UPDATE)));

        CreateTopicRequestHeader createTopicRequestHeader = new CreateTopicRequestHeader();
        createTopicRequestHeader.setTopic("topic");
        createTopicRequestHeader.setTopicFilterType(TopicFilterType.SINGLE_TAG.name());
        request = RemotingCommand.createRequestCommand(RequestCode.UPDATE_AND_CREATE_TOPIC, createTopicRequestHeader);
        request.setVersion(441);
        request.addExtField("AccessKey", "rocketmq");
        request.makeCustomHeaderToNet();
        result = builder.build(channelHandlerContext, request);
        Assert.assertEquals(1, result.size());
        Assert.assertEquals("User:rocketmq", result.get(0).getSubject().getSubjectKey());
        Assert.assertEquals("Topic:topic", result.get(0).getResource().getResourceKey());
        Assert.assertTrue(result.get(0).getActions().containsAll(Arrays.asList(Action.CREATE)));

        CreateUserRequestHeader createUserRequestHeader = new CreateUserRequestHeader();
        createUserRequestHeader.setUsername("abc");
        request = RemotingCommand.createRequestCommand(RequestCode.AUTH_CREATE_USER, createUserRequestHeader);
        request.setVersion(441);
        request.addExtField("AccessKey", "rocketmq");
        request.makeCustomHeaderToNet();
        result = builder.build(channelHandlerContext, request);
        Assert.assertEquals(1, result.size());
        Assert.assertEquals("User:rocketmq", result.get(0).getSubject().getSubjectKey());
        Assert.assertEquals("Cluster:DefaultCluster", result.get(0).getResource().getResourceKey());
        Assert.assertTrue(result.get(0).getActions().containsAll(Arrays.asList(Action.UPDATE)));
    }