public void receiveMessage(ProxyContext ctx, ReceiveMessageRequest request,
        StreamObserver<ReceiveMessageResponse> responseObserver) {
        ReceiveMessageResponseStreamWriter writer = createWriter(ctx, responseObserver);

        try {
            Settings settings = this.grpcClientSettingsManager.getClientSettings(ctx);
            Subscription subscription = settings.getSubscription();
            boolean fifo = subscription.getFifo();
            int maxAttempts = settings.getBackoffPolicy().getMaxAttempts();
            ProxyConfig config = ConfigurationManager.getProxyConfig();

            Long timeRemaining = ctx.getRemainingMs();
            long pollingTime;
            if (request.hasLongPollingTimeout()) {
                pollingTime = Durations.toMillis(request.getLongPollingTimeout());
            } else {
                pollingTime = timeRemaining - Durations.toMillis(settings.getRequestTimeout()) / 2;
            }
            if (pollingTime < config.getGrpcClientConsumerMinLongPollingTimeoutMillis()) {
                pollingTime = config.getGrpcClientConsumerMinLongPollingTimeoutMillis();
            }
            if (pollingTime > config.getGrpcClientConsumerMaxLongPollingTimeoutMillis()) {
                pollingTime = config.getGrpcClientConsumerMaxLongPollingTimeoutMillis();
            }

            if (pollingTime > timeRemaining) {
                if (timeRemaining >= config.getGrpcClientConsumerMinLongPollingTimeoutMillis()) {
                    pollingTime = timeRemaining;
                } else {
                    final String clientVersion = ctx.getClientVersion();
                    Code code =
                        null == clientVersion || ILLEGAL_POLLING_TIME_INTRODUCED_CLIENT_VERSION.compareTo(clientVersion) > 0 ?
                        Code.BAD_REQUEST : Code.ILLEGAL_POLLING_TIME;
                    writer.writeAndComplete(ctx, code, "The deadline time remaining is not enough" +
                        " for polling, please check network condition");
                    return;
                }
            }

            validateTopicAndConsumerGroup(request.getMessageQueue().getTopic(), request.getGroup());
            String topic = request.getMessageQueue().getTopic().getName();
            String group = request.getGroup().getName();

            long actualInvisibleTime = Durations.toMillis(request.getInvisibleDuration());
            ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
            if (proxyConfig.isEnableProxyAutoRenew() && request.getAutoRenew()) {
                actualInvisibleTime = proxyConfig.getDefaultInvisibleTimeMills();
            } else {
                validateInvisibleTime(actualInvisibleTime,
                    ConfigurationManager.getProxyConfig().getMinInvisibleTimeMillsForRecv());
            }

            FilterExpression filterExpression = request.getFilterExpression();
            SubscriptionData subscriptionData;
            try {
                subscriptionData = FilterAPI.build(topic, filterExpression.getExpression(),
                    GrpcConverter.getInstance().buildExpressionType(filterExpression.getType()));
            } catch (Exception e) {
                writer.writeAndComplete(ctx, Code.ILLEGAL_FILTER_EXPRESSION, e.getMessage());
                return;
            }

            this.messagingProcessor.popMessage(
                    ctx,
                    new ReceiveMessageQueueSelector(
                        request.getMessageQueue().getBroker().getName()
                    ),
                    group,
                    topic,
                    request.getBatchSize(),
                    actualInvisibleTime,
                    pollingTime,
                    ConsumeInitMode.MAX,
                    subscriptionData,
                    fifo,
                    new PopMessageResultFilterImpl(maxAttempts),
                    request.hasAttemptId() ? request.getAttemptId() : null,
                    timeRemaining
                ).thenAccept(popResult -> {
                    if (proxyConfig.isEnableProxyAutoRenew() && request.getAutoRenew()) {
                        if (PopStatus.FOUND.equals(popResult.getPopStatus())) {
                            List<MessageExt> messageExtList = popResult.getMsgFoundList();
                            for (MessageExt messageExt : messageExtList) {
                                String receiptHandle = messageExt.getProperty(MessageConst.PROPERTY_POP_CK);
                                if (receiptHandle != null) {
                                    MessageReceiptHandle messageReceiptHandle =
                                        new MessageReceiptHandle(group, topic, messageExt.getQueueId(), receiptHandle, messageExt.getMsgId(),
                                            messageExt.getQueueOffset(), messageExt.getReconsumeTimes());
                                    messagingProcessor.addReceiptHandle(ctx, grpcChannelManager.getChannel(ctx.getClientID()), group, messageExt.getMsgId(), messageReceiptHandle);
                                }
                            }
                        }
                    }
                    writer.writeAndComplete(ctx, request, popResult);
                })
                .exceptionally(t -> {
                    writer.writeAndComplete(ctx, request, t);
                    return null;
                });
        } catch (Throwable t) {
            writer.writeAndComplete(ctx, request, t);
        }
    }
@Test
    public void testReceiveMessageIllegalInvisibleTimeTooSmall() {
        StreamObserver<ReceiveMessageResponse> receiveStreamObserver = mock(ServerCallStreamObserver.class);
        ArgumentCaptor<ReceiveMessageResponse> responseArgumentCaptor = ArgumentCaptor.forClass(ReceiveMessageResponse.class);
        doNothing().when(receiveStreamObserver).onNext(responseArgumentCaptor.capture());

        when(this.grpcClientSettingsManager.getClientSettings(any())).thenReturn(Settings.newBuilder().getDefaultInstanceForType());

        this.receiveMessageActivity.receiveMessage(
            createContext(),
            ReceiveMessageRequest.newBuilder()
                .setGroup(Resource.newBuilder().setName(CONSUMER_GROUP).build())
                .setMessageQueue(MessageQueue.newBuilder().setTopic(Resource.newBuilder().setName(TOPIC).build()).build())
                .setAutoRenew(false)
                .setInvisibleDuration(Durations.fromSeconds(0))
                .build(),
            receiveStreamObserver
        );

        assertEquals(Code.ILLEGAL_INVISIBLE_TIME, getResponseCodeFromReceiveMessageResponseList(responseArgumentCaptor.getAllValues()));
    }