public void writeAndComplete(ProxyContext ctx, ReceiveMessageRequest request, PopResult popResult) {
        PopStatus status = popResult.getPopStatus();
        List<MessageExt> messageFoundList = popResult.getMsgFoundList();
        try {
            switch (status) {
                case FOUND:
                    if (messageFoundList.isEmpty()) {
                        streamObserver.onNext(ReceiveMessageResponse.newBuilder()
                            .setStatus(ResponseBuilder.getInstance().buildStatus(Code.MESSAGE_NOT_FOUND, "no match message"))
                            .build());
                    } else {
                        streamObserver.onNext(ReceiveMessageResponse.newBuilder()
                            .setStatus(ResponseBuilder.getInstance().buildStatus(Code.OK, Code.OK.name()))
                            .build());
                        Iterator<MessageExt> messageIterator = messageFoundList.iterator();
                        while (messageIterator.hasNext()) {
                            MessageExt curMessageExt = messageIterator.next();
                            Message curMessage = convertToMessage(curMessageExt);
                            try {
                                streamObserver.onNext(ReceiveMessageResponse.newBuilder()
                                    .setMessage(curMessage)
                                    .build());
                            } catch (Throwable t) {
                                this.processThrowableWhenWriteMessage(t, ctx, request, curMessageExt);
                                messageIterator.forEachRemaining(messageExt ->
                                    this.processThrowableWhenWriteMessage(t, ctx, request, messageExt));
                                return;
                            }
                        }
                    }
                    break;
                case POLLING_FULL:
                    streamObserver.onNext(ReceiveMessageResponse.newBuilder()
                        .setStatus(ResponseBuilder.getInstance().buildStatus(Code.TOO_MANY_REQUESTS, "polling full"))
                        .build());
                    break;
                case NO_NEW_MSG:
                case POLLING_NOT_FOUND:
                default:
                    streamObserver.onNext(ReceiveMessageResponse.newBuilder()
                        .setStatus(ResponseBuilder.getInstance().buildStatus(Code.MESSAGE_NOT_FOUND, "no new message"))
                        .build());
                    break;
            }
        } catch (Throwable t) {
            writeResponseWithErrorIgnore(
                ReceiveMessageResponse.newBuilder().setStatus(ResponseBuilder.getInstance().buildStatus(t)).build());
        } finally {
            onComplete();
        }
    }
@Test
    public void testWriteMessage() {
        ArgumentCaptor<String> changeInvisibleTimeMsgIdCaptor = ArgumentCaptor.forClass(String.class);
        doReturn(CompletableFuture.completedFuture(mock(AckResult.class))).when(this.messagingProcessor)
            .changeInvisibleTime(any(), any(), changeInvisibleTimeMsgIdCaptor.capture(), anyString(), anyString(), anyLong());

        ArgumentCaptor<ReceiveMessageResponse> responseArgumentCaptor = ArgumentCaptor.forClass(ReceiveMessageResponse.class);
        AtomicInteger onNextCallNum = new AtomicInteger(0);
        doAnswer(mock -> {
            if (onNextCallNum.incrementAndGet() > 2) {
                throw new RuntimeException();
            }
            return null;
        }).when(streamObserver).onNext(responseArgumentCaptor.capture());

        List<MessageExt> messageExtList = new ArrayList<>();
        messageExtList.add(createMessageExt(TOPIC, "tag"));
        messageExtList.add(createMessageExt(TOPIC, "tag"));
        PopResult popResult = new PopResult(PopStatus.FOUND, messageExtList);
        writer.writeAndComplete(
            ProxyContext.create(),
            ReceiveMessageRequest.newBuilder()
                .setGroup(Resource.newBuilder().setName(CONSUMER_GROUP).build())
                .setMessageQueue(MessageQueue.newBuilder().setTopic(Resource.newBuilder().setName(TOPIC).build()).build())
                .setFilterExpression(FilterExpression.newBuilder()
                    .setType(FilterType.TAG)
                    .setExpression("*")
                    .build())
                .build(),
            popResult
        );

        verify(streamObserver, times(1)).onCompleted();
        verify(streamObserver, times(4)).onNext(any());
        verify(this.messagingProcessor, times(1))
            .changeInvisibleTime(any(), any(), anyString(), anyString(), anyString(), anyLong());

        assertTrue(responseArgumentCaptor.getAllValues().get(0).hasStatus());
        assertEquals(Code.OK, responseArgumentCaptor.getAllValues().get(0).getStatus().getCode());
        assertTrue(responseArgumentCaptor.getAllValues().get(1).hasMessage());
        assertEquals(messageExtList.get(0).getMsgId(), responseArgumentCaptor.getAllValues().get(1).getMessage().getSystemProperties().getMessageId());

        assertEquals(messageExtList.get(1).getMsgId(), changeInvisibleTimeMsgIdCaptor.getValue());
    }