opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-10-29 17:25:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #3949] setDeliveryTimestamp when complete ReceiveMessageResponse

Browse Source
This commit is contained in:
kaiyi.lk
2022-07-13 11:30:49 +08:00
committed by zhouxiang
parent 74b8084268
commit e21c5c06b0
6 changed files with 86 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/ResponseWriter.java
+19 -28
View File
@@ -17,9 +17,10 @@
package org.apache.rocketmq.proxy.grpc.v2.common;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.stub.ServerCallStreamObserver;
import io.grpc.stub.StreamObserver;
import java.util.Iterator;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
@@ -27,41 +28,31 @@ import org.apache.rocketmq.logging.InternalLoggerFactory;
public class ResponseWriter {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
public static <T> void write(StreamObserver<T> observer, final Iterator<T> response) {
while (response.hasNext()) {
writeResponse(observer, response.next());
}
observer.onCompleted();
}
public static <T> void write(StreamObserver<T> observer, final T response) {
writeResponse(observer, response);
observer.onCompleted();
if (writeResponse(observer, response)) {
observer.onCompleted();
}
}
public static <T> void writeResponse(StreamObserver<T> observer, final T response) {
public static <T> boolean writeResponse(StreamObserver<T> observer, final T response) {
if (null == response) {
return;
}
if (isCancelled(observer)) {
log.warn("client has cancelled the request. response to write: {}", response);
return;
return false;
}
log.debug("start to write response. response: {}", response);
observer.onNext(response);
}
public static <T> void writeException(StreamObserver<T> observer, final Throwable e) {
if (null == e) {
return;
}
if (isCancelled(observer)) {
log.warn("Client has cancelled the request. Exception to write", e);
return;
log.warn("client has cancelled the request. response to write: {}", response);
return false;
}
log.debug("Start to write error response", e);
observer.onError(e);
observer.onCompleted();
try {
observer.onNext(response);
} catch (StatusRuntimeException statusRuntimeException) {
if (Status.CANCELLED.equals(statusRuntimeException.getStatus())) {
log.warn("client has cancelled the request. response to write: {}", response);
return false;
}
throw statusRuntimeException;
}
return true;
}
public static <T> boolean isCancelled(StreamObserver<T> observer) {
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivity.java
+13 -9
View File
@@ -51,10 +51,7 @@ public class ReceiveMessageActivity extends AbstractMessingActivity {
public void receiveMessage(Context ctx, ReceiveMessageRequest request,
StreamObserver<ReceiveMessageResponse> responseObserver) {
ProxyContext proxyContext = createContext(ctx);
ReceiveMessageResponseStreamWriter writer = new ReceiveMessageResponseStreamWriter(
this.messagingProcessor,
responseObserver
);
ReceiveMessageResponseStreamWriter writer = createWriter(proxyContext, responseObserver);
try {
Settings settings = this.grpcClientSettingsManager.getClientSettings(proxyContext);
@@ -67,7 +64,7 @@ public class ReceiveMessageActivity extends AbstractMessingActivity {
pollTime = timeRemaining;
}
if (pollTime <= 0) {
writer.write(proxyContext, Code.MESSAGE_NOT_FOUND, "time remaining is too small");
writer.writeAndComplete(proxyContext, Code.MESSAGE_NOT_FOUND, "time remaining is too small");
return;
}
@@ -85,7 +82,7 @@ public class ReceiveMessageActivity extends AbstractMessingActivity {
subscriptionData = FilterAPI.build(topic, filterExpression.getExpression(),
GrpcConverter.buildExpressionType(filterExpression.getType()));
} catch (Exception e) {
writer.write(proxyContext, Code.ILLEGAL_FILTER_EXPRESSION, e.getMessage());
writer.writeAndComplete(proxyContext, Code.ILLEGAL_FILTER_EXPRESSION, e.getMessage());
return;
}
@@ -104,16 +101,23 @@ public class ReceiveMessageActivity extends AbstractMessingActivity {
fifo,
new PopMessageResultFilterImpl(grpcClientSettingsManager),
timeRemaining
).thenAccept(popResult -> writer.write(proxyContext, request, popResult))
).thenAccept(popResult -> writer.writeAndComplete(proxyContext, request, popResult))
.exceptionally(t -> {
writer.write(proxyContext, request, t);
writer.writeAndComplete(proxyContext, request, t);
return null;
});
} catch (Throwable t) {
writer.write(proxyContext, request, t);
writer.writeAndComplete(proxyContext, request, t);
}
}
protected ReceiveMessageResponseStreamWriter createWriter(ProxyContext ctx, StreamObserver<ReceiveMessageResponse> responseObserver) {
return new ReceiveMessageResponseStreamWriter(
this.messagingProcessor,
responseObserver
);
}
protected static class ReceiveMessageQueueSelector implements QueueSelector {
private final String brokerName;
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageResponseStreamWriter.java
+37 -15
View File
@@ -20,14 +20,18 @@ import apache.rocketmq.v2.Code;
import apache.rocketmq.v2.Message;
import apache.rocketmq.v2.ReceiveMessageRequest;
import apache.rocketmq.v2.ReceiveMessageResponse;
import com.google.protobuf.util.Timestamps;
import io.grpc.stub.StreamObserver;
import java.time.Duration;
import java.util.Iterator;
import java.util.List;
import org.apache.rocketmq.client.consumer.PopResult;
import org.apache.rocketmq.client.consumer.PopStatus;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.consumer.ReceiptHandle;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter;
import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder;
@@ -35,7 +39,7 @@ import org.apache.rocketmq.proxy.grpc.v2.common.ResponseWriter;
import org.apache.rocketmq.proxy.processor.MessagingProcessor;
public class ReceiveMessageResponseStreamWriter {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
protected static final long NACK_INVISIBLE_TIME = Duration.ofSeconds(1).toMillis();
protected final MessagingProcessor messagingProcessor;
@@ -48,7 +52,7 @@ public class ReceiveMessageResponseStreamWriter {
this.streamObserver = observer;
}
public void write(ProxyContext ctx, ReceiveMessageRequest request, PopResult popResult) {
public void writeAndComplete(ProxyContext ctx, ReceiveMessageRequest request, PopResult popResult) {
PopStatus status = popResult.getPopStatus();
List<MessageExt> messageFoundList = popResult.getMsgFoundList();
try {
@@ -62,7 +66,7 @@ public class ReceiveMessageResponseStreamWriter {
streamObserver.onNext(ReceiveMessageResponse.newBuilder()
.setStatus(ResponseBuilder.buildStatus(Code.OK, Code.OK.name()))
.build());
Iterator<MessageExt> messageIterator = messageFoundList.iterator();
Iterator<MessageExt> messageIterator = messageFoundList.iterator();
while (messageIterator.hasNext()) {
MessageExt curMessageExt = messageIterator.next();
Message curMessage = convertToMessage(curMessageExt);
@@ -93,9 +97,10 @@ public class ReceiveMessageResponseStreamWriter {
break;
}
} catch (Throwable t) {
write(ctx, request, t);
writeResponseWithErrorIgnore(
ReceiveMessageResponse.newBuilder().setStatus(ResponseBuilder.buildStatus(t)).build());
} finally {
streamObserver.onCompleted();
onComplete();
}
}
@@ -121,17 +126,34 @@ public class ReceiveMessageResponseStreamWriter {
);
}
public void write(ProxyContext ctx, Code code, String message) {
ResponseWriter.write(
streamObserver,
ReceiveMessageResponse.newBuilder().setStatus(ResponseBuilder.buildStatus(code, message)).build()
);
public void writeAndComplete(ProxyContext ctx, Code code, String message) {
writeResponseWithErrorIgnore(
ReceiveMessageResponse.newBuilder().setStatus(ResponseBuilder.buildStatus(code, message)).build());
onComplete();
}
public void write(ProxyContext ctx, ReceiveMessageRequest request, Throwable throwable) {
ResponseWriter.write(
streamObserver,
ReceiveMessageResponse.newBuilder().setStatus(ResponseBuilder.buildStatus(throwable)).build()
);
public void writeAndComplete(ProxyContext ctx, ReceiveMessageRequest request, Throwable throwable) {
writeResponseWithErrorIgnore(
ReceiveMessageResponse.newBuilder().setStatus(ResponseBuilder.buildStatus(throwable)).build());
onComplete();
}
protected void writeResponseWithErrorIgnore(ReceiveMessageResponse response) {
try {
ResponseWriter.writeResponse(streamObserver, response);
} catch (Exception e) {
log.error("err when write receive message response", e);
}
}
protected void onComplete() {
writeResponseWithErrorIgnore(ReceiveMessageResponse.newBuilder()
.setDeliveryTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
.build());
try {
streamObserver.onCompleted();
} catch (Exception e) {
log.error("err when complete receive message response", e);
}
}
}
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivityTest.java
+11 -2
View File
@@ -94,7 +94,7 @@ public class ReceiveMessageActivityTest extends BaseActivityTest {
receiveStreamObserver
);
assertEquals(Code.ILLEGAL_FILTER_EXPRESSION, responseArgumentCaptor.getValue().getStatus().getCode());
assertEquals(Code.ILLEGAL_FILTER_EXPRESSION, getResponseCodeFromReceiveMessageResponseList(responseArgumentCaptor.getAllValues()));
}
@Test
@@ -133,7 +133,16 @@ public class ReceiveMessageActivityTest extends BaseActivityTest {
.build(),
receiveStreamObserver
);
assertEquals(Code.MESSAGE_NOT_FOUND, responseArgumentCaptor.getValue().getStatus().getCode());
assertEquals(Code.MESSAGE_NOT_FOUND, getResponseCodeFromReceiveMessageResponseList(responseArgumentCaptor.getAllValues()));
}
private Code getResponseCodeFromReceiveMessageResponseList(List<ReceiveMessageResponse> responseList) {
for (ReceiveMessageResponse response : responseList) {
if (response.hasStatus()) {
return response.getStatus().getCode();
}
}
return null;
}
@Test
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageResponseStreamWriterTest.java
+6 -4
View File
@@ -90,7 +90,7 @@ public class ReceiveMessageResponseStreamWriterTest extends BaseActivityTest {
messageExtList.add(createMessageExt(TOPIC, "tag"));
messageExtList.add(createMessageExt(TOPIC, "tag"));
PopResult popResult = new PopResult(PopStatus.FOUND, messageExtList);
writer.write(
writer.writeAndComplete(
ProxyContext.create(),
ReceiveMessageRequest.newBuilder()
.setGroup(Resource.newBuilder().setName(CONSUMER_GROUP).build())
@@ -104,7 +104,7 @@ public class ReceiveMessageResponseStreamWriterTest extends BaseActivityTest {
);
verify(streamObserver, times(1)).onCompleted();
verify(streamObserver, times(3)).onNext(any());
verify(streamObserver, times(4)).onNext(any());
verify(this.messagingProcessor, times(1))
.changeInvisibleTime(any(), any(), anyString(), anyString(), anyString(), anyLong());
@@ -122,7 +122,7 @@ public class ReceiveMessageResponseStreamWriterTest extends BaseActivityTest {
doNothing().when(streamObserver).onNext(responseArgumentCaptor.capture());
PopResult popResult = new PopResult(PopStatus.POLLING_FULL, new ArrayList<>());
writer.write(
writer.writeAndComplete(
ProxyContext.create(),
ReceiveMessageRequest.newBuilder()
.setGroup(Resource.newBuilder().setName(CONSUMER_GROUP).build())
@@ -135,7 +135,9 @@ public class ReceiveMessageResponseStreamWriterTest extends BaseActivityTest {
popResult
);
assertEquals(Code.TOO_MANY_REQUESTS, responseArgumentCaptor.getValue().getStatus().getCode());
ReceiveMessageResponse response = responseArgumentCaptor.getAllValues().stream().filter(ReceiveMessageResponse::hasStatus)
.findFirst().get();
assertEquals(Code.TOO_MANY_REQUESTS, response.getStatus().getCode());
}
private static MessageExt createMessageExt(String topic, String tags) {