This is an automated email from the ASF dual-hosted git repository.
mxsm 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 800b628af [ISSUE #4264]Some optimizations for ProducerService (#4334)
800b628af is described below
commit 800b628af50b2b6d3a7401b81d45506498673608
Author: zhengpeng <[email protected]>
AuthorDate: Sat Aug 5 20:27:31 2023 +0800
[ISSUE #4264]Some optimizations for ProducerService (#4334)
---
.../eventmesh/connector/openfunction/service/ConsumerService.java | 6 +++---
.../eventmesh/connector/openfunction/service/ProducerService.java | 8 ++++----
2 files changed, 7 insertions(+), 7 deletions(-)
diff --git
a/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/service/ConsumerService.java
b/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/service/ConsumerService.java
index 1df20652a..26baa4ab1 100644
---
a/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/service/ConsumerService.java
+++
b/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/service/ConsumerService.java
@@ -42,11 +42,11 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
public class ConsumerService extends
ConsumerServiceGrpc.ConsumerServiceImplBase {
- public OpenFunctionSinkConnector openFunctionSinkConnector;
+ private final OpenFunctionSinkConnector openFunctionSinkConnector;
- public BlockingQueue<ConnectRecord> queue;
+ private final BlockingQueue<ConnectRecord> queue;
- public OpenFunctionServerConfig config;
+ private final OpenFunctionServerConfig config;
private final transient ManagedChannel channel =
ManagedChannelBuilder.forAddress("127.0.0.1", 10115).usePlaintext().build();
diff --git
a/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/service/ProducerService.java
b/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/service/ProducerService.java
index 42ccea042..36d58ceb6 100644
---
a/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/service/ProducerService.java
+++
b/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/service/ProducerService.java
@@ -44,11 +44,11 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
public class ProducerService extends
PublisherServiceGrpc.PublisherServiceImplBase {
- public OpenFunctionSourceConnector openFunctionSourceConnector;
+ private final OpenFunctionSourceConnector openFunctionSourceConnector;
- public BlockingQueue<ConnectRecord> queue;
+ private final BlockingQueue<ConnectRecord> queue;
- public OpenFunctionServerConfig config;
+ private final OpenFunctionServerConfig config;
public ProducerService(OpenFunctionSourceConnector
openFunctionSourceConnector, OpenFunctionServerConfig serverConfig) {
this.openFunctionSourceConnector = openFunctionSourceConnector;
@@ -85,7 +85,7 @@ public class ProducerService extends
PublisherServiceGrpc.PublisherServiceImplBa
CloudEventAttributeValue.newBuilder().setCeString(StatusCode.EVENTMESH_SEND_ASYNC_MSG_ERR.getErrMsg()).build())
.putAttributes(ProtocolKey.GRPC_RESPONSE_TIME,
CloudEventAttributeValue.newBuilder()
.setCeTimestamp(Timestamp.newBuilder().setSeconds(instant.getEpochSecond()).setNanos(instant.getNano()).build()).build());
- throw new RuntimeException(e);
+ Thread.currentThread().interrupt();
}
responseObserver.onNext(builder.build());
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
