This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 7f05cfbad [ISSUE #3815]DelayRetryable and RetryContext can be
refactored. (#3819)
7f05cfbad is described below
commit 7f05cfbad40255ed2f33a8ac74f491c4219841ee
Author: pandaapo <[email protected]>
AuthorDate: Sat May 6 16:31:53 2023 +0800
[ISSUE #3815]DelayRetryable and RetryContext can be refactored. (#3819)
* Refacotr DelayRetryable and RetryContext.
* Modify as review.
---
.../protocol/{grpc/retry => }/DelayRetryable.java | 6 +--
.../client/session/retry => }/RetryContext.java | 9 ++--
.../protocol/grpc/producer/SendMessageContext.java | 11 ++---
.../protocol/grpc/push/AbstractPushRequest.java | 5 +--
.../core/protocol/grpc/retry/GrpcRetryer.java | 1 +
.../core/protocol/grpc/retry/RetryContext.java | 50 ----------------------
.../protocol/http/producer/SendMessageContext.java | 12 +++---
.../http/push/AbstractHTTPPushRequest.java | 2 +-
.../protocol/http/push/AsyncHTTPPushRequest.java | 3 +-
.../core/protocol/http/retry/DelayRetryable.java | 28 ------------
.../core/protocol/http/retry/HttpRetryer.java | 1 +
.../core/protocol/http/retry/RetryContext.java | 50 ----------------------
.../client/session/push/DownStreamMsgContext.java | 2 +-
.../client/session/retry/EventMeshTcpRetryer.java | 10 ++++-
.../client/session/send/UpStreamMsgContext.java | 2 +-
15 files changed, 35 insertions(+), 157 deletions(-)
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/retry/DelayRetryable.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/DelayRetryable.java
similarity index 89%
rename from
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/retry/DelayRetryable.java
rename to
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/DelayRetryable.java
index a9486030b..b60876a64 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/retry/DelayRetryable.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/DelayRetryable.java
@@ -15,14 +15,14 @@
* limitations under the License.
*/
-package org.apache.eventmesh.runtime.core.protocol.grpc.retry;
+package org.apache.eventmesh.runtime.core.protocol;
import java.util.concurrent.Delayed;
/**
- *
+ * Retry
*/
public interface DelayRetryable extends Delayed {
- boolean retry() throws Exception;
+ void retry() throws Exception;
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/retry/RetryContext.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/RetryContext.java
similarity index 88%
rename from
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/retry/RetryContext.java
rename to
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/RetryContext.java
index 5a7201387..939ef40dd 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/retry/RetryContext.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/RetryContext.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.eventmesh.runtime.core.protocol.tcp.client.session.retry;
+package org.apache.eventmesh.runtime.core.protocol;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
@@ -24,14 +24,13 @@ import javax.annotation.Nonnull;
import io.cloudevents.CloudEvent;
-
-public abstract class RetryContext implements Delayed {
+public abstract class RetryContext implements DelayRetryable {
public CloudEvent event;
public String seq;
- public int retryTimes = 0;
+ public int retryTimes;
public long executeTime = System.currentTimeMillis();
@@ -51,6 +50,4 @@ public abstract class RetryContext implements Delayed {
public long getDelay(TimeUnit unit) {
return unit.convert(this.executeTime - System.currentTimeMillis(),
TimeUnit.MILLISECONDS);
}
-
- public abstract void retry();
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/producer/SendMessageContext.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/producer/SendMessageContext.java
index 6d33129c0..dcc87248c 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/producer/SendMessageContext.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/producer/SendMessageContext.java
@@ -22,7 +22,7 @@ import org.apache.eventmesh.api.SendResult;
import org.apache.eventmesh.api.exception.OnExceptionContext;
import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.runtime.boot.EventMeshGrpcServer;
-import org.apache.eventmesh.runtime.core.protocol.grpc.retry.RetryContext;
+import org.apache.eventmesh.runtime.core.protocol.RetryContext;
import org.apache.commons.lang3.time.DateFormatUtils;
@@ -101,13 +101,15 @@ public class SendMessageContext extends RetryContext {
}
@Override
- public boolean retry() throws Exception {
+ public void retry() throws Exception {
if (eventMeshProducer == null) {
- return false;
+ logger.error("Exception happends during retry. EventMeshProducer
is null.");
+ return;
}
if (retryTimes > 0) { //retry once
- return false;
+ logger.error("Exception happends during retry. The retryTimes >
0.");
+ return;
}
retryTimes++;
@@ -122,6 +124,5 @@ public class SendMessageContext extends RetryContext {
logger.warn("", context.getException());
}
});
- return true;
}
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/push/AbstractPushRequest.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/push/AbstractPushRequest.java
index d8170046a..c3a0067f4 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/push/AbstractPushRequest.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/push/AbstractPushRequest.java
@@ -28,9 +28,9 @@ import
org.apache.eventmesh.protocol.api.ProtocolPluginFactory;
import org.apache.eventmesh.runtime.boot.EventMeshGrpcServer;
import org.apache.eventmesh.runtime.configuration.EventMeshGrpcConfiguration;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import org.apache.eventmesh.runtime.core.protocol.RetryContext;
import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.EventMeshConsumer;
import org.apache.eventmesh.runtime.core.protocol.grpc.retry.GrpcRetryer;
-import org.apache.eventmesh.runtime.core.protocol.grpc.retry.RetryContext;
import java.util.Collections;
import java.util.Map;
@@ -101,9 +101,8 @@ public abstract class AbstractPushRequest extends
RetryContext {
}
@Override
- public boolean retry() {
+ public void retry() {
tryPushRequest();
- return true;
}
protected void delayRetry() {
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/retry/GrpcRetryer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/retry/GrpcRetryer.java
index 034b0db1d..4c5cea103 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/retry/GrpcRetryer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/retry/GrpcRetryer.java
@@ -20,6 +20,7 @@ package org.apache.eventmesh.runtime.core.protocol.grpc.retry;
import org.apache.eventmesh.common.EventMeshThreadFactory;
import org.apache.eventmesh.runtime.boot.EventMeshGrpcServer;
import org.apache.eventmesh.runtime.configuration.EventMeshGrpcConfiguration;
+import org.apache.eventmesh.runtime.core.protocol.DelayRetryable;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.DelayQueue;
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/retry/RetryContext.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/retry/RetryContext.java
deleted file mode 100644
index c09f3b9f5..000000000
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/retry/RetryContext.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.eventmesh.runtime.core.protocol.grpc.retry;
-
-import java.util.concurrent.Delayed;
-import java.util.concurrent.TimeUnit;
-
-public abstract class RetryContext implements DelayRetryable {
-
- public int retryTimes = 0;
-
- public long executeTime = System.currentTimeMillis();
-
- public RetryContext delay(long delay) {
- this.executeTime = System.currentTimeMillis() + delay;
- return this;
- }
-
- @Override
- public int compareTo(Delayed delayed) {
- RetryContext obj = (RetryContext) delayed;
- if (this.executeTime > obj.executeTime) {
- return 1;
- } else if (this.executeTime == obj.executeTime) {
- return 0;
- } else {
- return -1;
- }
- }
-
- @Override
- public long getDelay(TimeUnit unit) {
- return unit.convert(this.executeTime - System.currentTimeMillis(),
TimeUnit.MILLISECONDS);
- }
-}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/producer/SendMessageContext.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/producer/SendMessageContext.java
index c072b60a1..3334c26bd 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/producer/SendMessageContext.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/producer/SendMessageContext.java
@@ -22,7 +22,7 @@ import org.apache.eventmesh.api.SendResult;
import org.apache.eventmesh.api.exception.OnExceptionContext;
import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
-import org.apache.eventmesh.runtime.core.protocol.http.retry.RetryContext;
+import org.apache.eventmesh.runtime.core.protocol.RetryContext;
import org.apache.commons.lang3.time.DateFormatUtils;
@@ -123,13 +123,15 @@ public class SendMessageContext extends RetryContext {
}
@Override
- public boolean retry() throws Exception {
+ public void retry() throws Exception {
if (eventMeshProducer == null) {
- return false;
+ log.error("Exception happends during retry. EventMeshProduceer is
null.");
+ return;
}
if (retryTimes > 0) { //retry once
- return false;
+ log.error("Exception happends during retry. The retryTimes > 0.");
+ return;
}
retryTimes++;
@@ -146,7 +148,5 @@ public class SendMessageContext extends RetryContext {
}
});
-
- return true;
}
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AbstractHTTPPushRequest.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AbstractHTTPPushRequest.java
index 3fd83268d..408d130b0 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AbstractHTTPPushRequest.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AbstractHTTPPushRequest.java
@@ -20,9 +20,9 @@ package org.apache.eventmesh.runtime.core.protocol.http.push;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
import org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import org.apache.eventmesh.runtime.core.protocol.RetryContext;
import
org.apache.eventmesh.runtime.core.protocol.http.consumer.HandleMsgContext;
import org.apache.eventmesh.runtime.core.protocol.http.retry.HttpRetryer;
-import org.apache.eventmesh.runtime.core.protocol.http.retry.RetryContext;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java
index 02732e6c3..f96e05f58 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java
@@ -366,8 +366,7 @@ public class AsyncHTTPPushRequest extends
AbstractHTTPPushRequest {
}
@Override
- public boolean retry() {
+ public void retry() {
tryHTTPRequest();
- return true;
}
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/retry/DelayRetryable.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/retry/DelayRetryable.java
deleted file mode 100644
index d9d04ce36..000000000
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/retry/DelayRetryable.java
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.eventmesh.runtime.core.protocol.http.retry;
-
-import java.util.concurrent.Delayed;
-
-/**
- * DelayRetryable
- */
-public interface DelayRetryable extends Delayed {
-
- boolean retry() throws Exception;
-}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/retry/HttpRetryer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/retry/HttpRetryer.java
index 384fa400e..4e1be246d 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/retry/HttpRetryer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/retry/HttpRetryer.java
@@ -19,6 +19,7 @@ package org.apache.eventmesh.runtime.core.protocol.http.retry;
import org.apache.eventmesh.common.EventMeshThreadFactory;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
+import org.apache.eventmesh.runtime.core.protocol.DelayRetryable;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.DelayQueue;
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/retry/RetryContext.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/retry/RetryContext.java
deleted file mode 100644
index 7841cca1e..000000000
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/retry/RetryContext.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.eventmesh.runtime.core.protocol.http.retry;
-
-import java.util.concurrent.Delayed;
-import java.util.concurrent.TimeUnit;
-
-public abstract class RetryContext implements DelayRetryable {
-
- public int retryTimes = 0;
-
- public long executeTime = System.currentTimeMillis();
-
- public RetryContext delay(long delay) {
- this.executeTime = System.currentTimeMillis() + delay;
- return this;
- }
-
- @Override
- public int compareTo(Delayed delayed) {
- RetryContext obj = (RetryContext) delayed;
- if (this.executeTime > obj.executeTime) {
- return 1;
- } else if (this.executeTime == obj.executeTime) {
- return 0;
- } else {
- return -1;
- }
- }
-
- @Override
- public long getDelay(TimeUnit unit) {
- return unit.convert(this.executeTime - System.currentTimeMillis(),
TimeUnit.MILLISECONDS);
- }
-}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/push/DownStreamMsgContext.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/push/DownStreamMsgContext.java
index abc328ef1..16a97d4cb 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/push/DownStreamMsgContext.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/push/DownStreamMsgContext.java
@@ -22,8 +22,8 @@ import org.apache.eventmesh.common.protocol.SubscriptionItem;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
import org.apache.eventmesh.runtime.core.plugin.MQConsumerWrapper;
+import org.apache.eventmesh.runtime.core.protocol.RetryContext;
import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
-import
org.apache.eventmesh.runtime.core.protocol.tcp.client.session.retry.RetryContext;
import org.apache.eventmesh.runtime.util.EventMeshUtil;
import org.apache.eventmesh.runtime.util.ServerGlobal;
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/retry/EventMeshTcpRetryer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/retry/EventMeshTcpRetryer.java
index 6fb155f34..a28ae09c4 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/retry/EventMeshTcpRetryer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/retry/EventMeshTcpRetryer.java
@@ -20,6 +20,7 @@ package
org.apache.eventmesh.runtime.core.protocol.tcp.client.session.retry;
import org.apache.eventmesh.common.EventMeshThreadFactory;
import org.apache.eventmesh.common.protocol.SubscriptionType;
import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
+import org.apache.eventmesh.runtime.core.protocol.RetryContext;
import
org.apache.eventmesh.runtime.core.protocol.tcp.client.session.push.DownStreamMsgContext;
import org.apache.eventmesh.runtime.util.EventMeshUtil;
@@ -91,7 +92,14 @@ public class EventMeshTcpRetryer {
try {
RetryContext retryContext;
while ((retryContext = retrys.take()) != null) {
- pool.execute(retryContext::retry);
+ final RetryContext retryCtx = retryContext;
+ pool.execute(() -> {
+ try {
+ retryCtx.retry();
+ } catch (Exception e) {
+ log.error("retry-dispatcher error!", e);
+ }
+ });
}
} catch (Exception e) {
log.error("retry-dispatcher error!", e);
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/send/UpStreamMsgContext.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/send/UpStreamMsgContext.java
index 78d6ba407..4ab4fd609 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/send/UpStreamMsgContext.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/send/UpStreamMsgContext.java
@@ -25,8 +25,8 @@ import org.apache.eventmesh.common.protocol.tcp.Header;
import org.apache.eventmesh.common.protocol.tcp.OPStatus;
import org.apache.eventmesh.common.protocol.tcp.Package;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import org.apache.eventmesh.runtime.core.protocol.RetryContext;
import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
-import
org.apache.eventmesh.runtime.core.protocol.tcp.client.session.retry.RetryContext;
import org.apache.eventmesh.runtime.util.EventMeshUtil;
import org.apache.eventmesh.runtime.util.Utils;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
