You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicecomb.apache.org by ni...@apache.org on 2018/09/21 14:33:56 UTC
[incubator-servicecomb-saga] branch master updated (24ca70b ->
5350a86)
This is an automated email from the ASF dual-hosted git repository.
ningjiang pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-saga.git.
from 24ca70b SCB-923 Clean up the alpha code
new f4579fe SCB-909 Integrate TccLoadBalance to omega-transaction && modify saga message interface.
new 6594776 SCB-909 clean up GrpcTccEventService.
new ddcc55d SCB-909 Adjust saga load balance sender package.
new fdfc6d7 SCB-909 Replace LoadBalancedClusterMessageSender to SagaLoadBalanceSender.
new cd148fd SCB-909 Optimize package of load balance.
new 5350a86 SCB-909 Revise copyright space.
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../omega/connector/grpc/AlphaClusterConfig.java | 1 -
.../omega/connector/grpc/GrpcTccEventService.java | 157 -------------
.../grpc/LoadBalancedClusterMessageSender.java | 242 ---------------------
.../connector/grpc/{ => core}/FastestSender.java | 20 +-
.../grpc/{tcc => core}/GrpcOnErrorHandler.java | 25 ++-
.../grpc/{tcc => core}/LoadBalanceContext.java | 15 +-
.../{tcc => core}/LoadBalanceContextBuilder.java | 35 +--
.../{tcc => core}/LoadBalanceSenderAdapter.java | 26 +--
.../grpc/{ => core}/MessageSenderPicker.java | 4 +-
.../grpc/{tcc => core}/PendingTaskRunner.java | 14 +-
.../grpc/{ => core}/PushBackReconnectRunnable.java | 2 +-
.../{tcc => core}/ReconnectStreamObserver.java | 15 +-
.../omega/connector/grpc/core/SenderExecutor.java | 8 +-
.../omega/connector/grpc/core/TransactionType.java | 8 +-
.../{ => saga}/GrpcCompensateStreamObserver.java | 37 ++--
.../GrpcSagaClientMessageSender.java} | 30 ++-
.../grpc/{ => saga}/RetryableMessageSender.java | 8 +-
.../grpc/{tcc => saga}/SagaLoadBalanceSender.java | 25 ++-
.../grpc/tcc/GrpcCoordinateStreamObserver.java | 14 +-
.../grpc/tcc/GrpcTccClientMessageSender.java | 20 +-
.../omega/connector/grpc/tcc/SenderExecutor.java | 25 ---
.../connector/grpc/tcc/TccLoadBalanceSender.java | 18 +-
...st.java => GrpcTccClientMessageSenderTest.java} | 7 +-
...LoadBalanceClusterMessageSenderWithTLSTest.java | 16 +-
.../grpc/LoadBalancedClusterMessageSenderTest.java | 50 +++--
.../LoadBalancedClusterMessageSenderTestBase.java | 8 +-
.../grpc/PushBackReconnectRunnableTest.java | 2 +-
.../connector/grpc/RetryableMessageSenderTest.java | 7 +-
.../grpc/tcc/LoadBalanceContextBuilderTest.java | 40 +++-
.../grpc/tcc/LoadBalanceSenderTestBase.java | 12 +-
.../connector/grpc/tcc/MyTccEventServiceImpl.java | 12 +-
.../grpc/tcc/SagaLoadBalanceSenderTest.java | 12 +-
.../grpc/tcc/TccLoadBalanceSenderTest.java | 29 ++-
.../saga/omega/spring/OmegaSpringConfig.java | 66 +++---
.../spring/TransactionAspectConfig.java | 26 ++-
.../omega/transaction/spring/MessageConfig.java | 10 +-
.../omega/transaction/CompensableInterceptor.java | 4 +-
.../transaction/CompensationMessageHandler.java | 4 +-
.../saga/omega/transaction/MessageSender.java | 14 +-
.../saga/omega/transaction/SagaMessageSender.java} | 7 +-
.../transaction/SagaStartAnnotationProcessor.java | 4 +-
.../saga/omega/transaction/SagaStartAspect.java | 2 +-
.../saga/omega/transaction/TransactionAspect.java | 2 +-
.../transaction/tcc/CoordinateMessageHandler.java | 8 +-
.../omega/transaction}/tcc/TccMessageSender.java | 2 +-
.../transaction/tcc/TccParticipatorAspect.java | 10 +-
.../tcc/TccStartAnnotationProcessor.java | 13 +-
.../saga/omega/transaction/tcc/TccStartAspect.java | 4 +-
.../transaction/CompensableInterceptorTest.java | 2 +-
.../CompensationMessageHandlerTest.java | 2 +-
.../omega/transaction/DefaultRecoveryTest.java | 4 +-
.../omega/transaction/ForwardRecoveryTest.java | 4 +-
.../SagaStartAnnotationProcessorTest.java | 4 +-
.../omega/transaction/SagaStartAspectTest.java | 3 +-
.../omega/transaction/TransactionAspectTest.java | 2 +-
.../tcc/CoordinateMessageHandlerTest.java | 6 +-
.../transaction/tcc/TccParticipatorAspectTest.java | 4 +-
.../tcc/TccStartAnnotationProcessorTest.java | 4 +-
.../omega/transaction/tcc/TccStartAspectTest.java | 4 +-
59 files changed, 386 insertions(+), 773 deletions(-)
delete mode 100644 omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcTccEventService.java
delete mode 100644 omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java
rename omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/{ => core}/FastestSender.java (61%)
rename omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/{tcc => core}/GrpcOnErrorHandler.java (73%)
rename omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/{tcc => core}/LoadBalanceContext.java (78%)
rename omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/{tcc => core}/LoadBalanceContextBuilder.java (83%)
rename omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/{tcc => core}/LoadBalanceSenderAdapter.java (80%)
rename omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/{ => core}/MessageSenderPicker.java (91%)
rename omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/{tcc => core}/PendingTaskRunner.java (78%)
rename omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/{ => core}/PushBackReconnectRunnable.java (97%)
rename omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/{tcc => core}/ReconnectStreamObserver.java (72%)
copy saga-core/src/main/java/org/apache/servicecomb/saga/core/EventStore.java => omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/SenderExecutor.java (80%)
copy alpha/alpha-core/src/main/java/org/apache/servicecomb/saga/alpha/core/TaskStatus.java => omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/TransactionType.java (88%)
rename omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/{ => saga}/GrpcCompensateStreamObserver.java (76%)
rename omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/{GrpcClientMessageSender.java => saga/GrpcSagaClientMessageSender.java} (87%)
rename omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/{ => saga}/RetryableMessageSender.java (87%)
rename omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/{tcc => saga}/SagaLoadBalanceSender.java (59%)
delete mode 100644 omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/SenderExecutor.java
rename omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/{GrpcTccEventServiceTest.java => GrpcTccClientMessageSenderTest.java} (97%)
rename omega/{omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TransactionType.java => omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/SagaMessageSender.java} (84%)
rename omega/{omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc => omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction}/tcc/TccMessageSender.java (96%)
[incubator-servicecomb-saga] 04/06: SCB-909 Replace
LoadBalancedClusterMessageSender to SagaLoadBalanceSender.
Posted by ni...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ningjiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-saga.git
commit fdfc6d760cd25db83a2cdf8e0c8d2a4b7b47fc10
Author: cherrylzhao <zh...@126.com>
AuthorDate: Fri Sep 21 17:05:06 2018 +0800
SCB-909 Replace LoadBalancedClusterMessageSender to SagaLoadBalanceSender.
---
.../grpc/saga/GrpcCompensateStreamObserver.java | 30 +++++-------
...ender.java => GrpcSagaClientMessageSender.java} | 15 +++---
.../saga/LoadBalancedClusterMessageSender.java | 20 ++++----
.../grpc/{tcc => saga}/SagaLoadBalanceSender.java | 5 +-
.../connector/grpc/tcc/LoadBalanceContext.java | 1 +
.../grpc/tcc/LoadBalanceContextBuilder.java | 20 ++++----
.../grpc/tcc/LoadBalanceSenderAdapter.java | 4 ++
...LoadBalanceClusterMessageSenderWithTLSTest.java | 16 +++---
.../grpc/LoadBalancedClusterMessageSenderTest.java | 39 ++++++++++-----
.../LoadBalancedClusterMessageSenderTestBase.java | 5 +-
.../saga/omega/spring/OmegaSpringConfig.java | 57 ++++++++++++++++------
11 files changed, 131 insertions(+), 81 deletions(-)
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcCompensateStreamObserver.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcCompensateStreamObserver.java
index f9a18b1..73bf0ff 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcCompensateStreamObserver.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcCompensateStreamObserver.java
@@ -19,28 +19,33 @@ package org.apache.servicecomb.saga.omega.connector.grpc.saga;
import java.lang.invoke.MethodHandles;
+import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContext;
+import org.apache.servicecomb.saga.omega.connector.grpc.tcc.ReconnectStreamObserver;
import org.apache.servicecomb.saga.omega.transaction.MessageDeserializer;
import org.apache.servicecomb.saga.omega.transaction.MessageHandler;
+import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcCompensateCommand;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.grpc.stub.StreamObserver;
-class GrpcCompensateStreamObserver implements StreamObserver<GrpcCompensateCommand> {
+class GrpcCompensateStreamObserver extends ReconnectStreamObserver<GrpcCompensateCommand> {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final MessageHandler messageHandler;
- private final Runnable errorHandler;
- private final MessageDeserializer deserializer;
-
- GrpcCompensateStreamObserver(MessageHandler messageHandler, Runnable errorHandler, MessageDeserializer deserializer) {
+ public GrpcCompensateStreamObserver(LoadBalanceContext loadContext,
+ MessageSender messageSender,
+ MessageHandler messageHandler, MessageDeserializer deserializer) {
+ super(loadContext, messageSender);
this.messageHandler = messageHandler;
- this.errorHandler = errorHandler;
this.deserializer = deserializer;
}
+ private final MessageHandler messageHandler;
+ private final MessageDeserializer deserializer;
+
+
@Override
public void onNext(GrpcCompensateCommand command) {
LOG.info("Received compensate command, global tx id: {}, local tx id: {}, compensation method: {}",
@@ -53,15 +58,4 @@ class GrpcCompensateStreamObserver implements StreamObserver<GrpcCompensateComma
command.getCompensationMethod(),
deserializer.deserialize(command.getPayloads().toByteArray()));
}
-
- @Override
- public void onError(Throwable t) {
- LOG.error("Failed to process grpc compensate command.", t);
- errorHandler.run();
- }
-
- @Override
- public void onCompleted() {
- // Do nothing here
- }
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcClientMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcSagaClientMessageSender.java
similarity index 91%
rename from omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcClientMessageSender.java
rename to omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcSagaClientMessageSender.java
index 86a59d7..61be7f4 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcClientMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcSagaClientMessageSender.java
@@ -19,7 +19,7 @@ package org.apache.servicecomb.saga.omega.connector.grpc.saga;
import com.google.protobuf.ByteString;
import io.grpc.ManagedChannel;
-import org.apache.servicecomb.saga.omega.connector.grpc.saga.LoadBalancedClusterMessageSender.ErrorHandlerFactory;
+import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContext;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.MessageDeserializer;
@@ -35,8 +35,9 @@ import org.apache.servicecomb.saga.pack.contract.grpc.TxEventServiceGrpc;
import org.apache.servicecomb.saga.pack.contract.grpc.TxEventServiceGrpc.TxEventServiceBlockingStub;
import org.apache.servicecomb.saga.pack.contract.grpc.TxEventServiceGrpc.TxEventServiceStub;
-public class GrpcClientMessageSender implements SagaMessageSender {
+public class GrpcSagaClientMessageSender implements SagaMessageSender {
private final String target;
+
private final TxEventServiceStub asyncEventService;
private final MessageSerializer serializer;
@@ -44,23 +45,23 @@ public class GrpcClientMessageSender implements SagaMessageSender {
private final TxEventServiceBlockingStub blockingEventService;
private final GrpcCompensateStreamObserver compensateStreamObserver;
+
private final GrpcServiceConfig serviceConfig;
- public GrpcClientMessageSender(
+ public GrpcSagaClientMessageSender(
String address,
ManagedChannel channel,
MessageSerializer serializer,
MessageDeserializer deserializer,
ServiceConfig serviceConfig,
- ErrorHandlerFactory errorHandlerFactory,
- MessageHandler handler) {
+ MessageHandler handler,
+ LoadBalanceContext loadContext) {
this.target = address;
this.asyncEventService = TxEventServiceGrpc.newStub(channel);
this.blockingEventService = TxEventServiceGrpc.newBlockingStub(channel);
this.serializer = serializer;
-
this.compensateStreamObserver =
- new GrpcCompensateStreamObserver(handler, errorHandlerFactory.getHandler(this), deserializer);
+ new GrpcCompensateStreamObserver(loadContext, this, handler, deserializer);
this.serviceConfig = serviceConfig(serviceConfig.serviceName(), serviceConfig.instanceId());
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/LoadBalancedClusterMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/LoadBalancedClusterMessageSender.java
index da02590..849125b 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/LoadBalancedClusterMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/LoadBalancedClusterMessageSender.java
@@ -106,16 +106,16 @@ public class LoadBalancedClusterMessageSender implements SagaMessageSender {
.build();
}
channels.add(channel);
- senders.put(
- new GrpcClientMessageSender(
- address,
- channel,
- clusterConfig.getMessageSerializer(),
- clusterConfig.getMessageDeserializer(),
- serviceConfig,
- new ErrorHandlerFactory(),
- clusterConfig.getMessageHandler()),
- 0L);
+// senders.put(
+// new GrpcSagaClientMessageSender(
+// address,
+// channel,
+// clusterConfig.getMessageSerializer(),
+// clusterConfig.getMessageDeserializer(),
+// serviceConfig,
+// new ErrorHandlerFactory(),
+// clusterConfig.getMessageHandler()),
+// 0L);
}
scheduleReconnectTask(reconnectDelay);
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/SagaLoadBalanceSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/SagaLoadBalanceSender.java
similarity index 86%
rename from omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/SagaLoadBalanceSender.java
rename to omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/SagaLoadBalanceSender.java
index ded20d7..119a2d3 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/SagaLoadBalanceSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/SagaLoadBalanceSender.java
@@ -15,10 +15,13 @@
* limitations under the License.
*/
-package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
+package org.apache.servicecomb.saga.omega.connector.grpc.saga;
import com.google.common.base.Optional;
import org.apache.servicecomb.saga.omega.connector.grpc.MessageSenderPicker;
+import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContext;
+import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceSenderAdapter;
+import org.apache.servicecomb.saga.omega.connector.grpc.tcc.SenderExecutor;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContext.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContext.java
index 3831a92..86c1991 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContext.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContext.java
@@ -57,6 +57,7 @@ public class LoadBalanceContext {
return grpcOnErrorHandler;
}
+ // this is only for test
public void setSenders(Map<MessageSender, Long> senders) {
this.senders = senders;
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilder.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilder.java
index 0b22ab6..d4411ec 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilder.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilder.java
@@ -36,6 +36,7 @@ import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import javax.net.ssl.SSLException;
import org.apache.servicecomb.saga.omega.connector.grpc.AlphaClusterConfig;
+import org.apache.servicecomb.saga.omega.connector.grpc.saga.GrpcSagaClientMessageSender;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
@@ -58,7 +59,6 @@ public class LoadBalanceContextBuilder {
}
public LoadBalanceContext build() {
-
if (clusterConfig.getAddresses().isEmpty()) {
throw new IllegalArgumentException("No reachable cluster address provided");
}
@@ -101,16 +101,16 @@ public class LoadBalanceContextBuilder {
address,
clusterConfig.getTccMessageHandler(),
loadContext);
-
case SAGA:
-// return new GrpcClientMessageSender(
-// address,
-// channel,
-// clusterConfig.getMessageSerializer(),
-// clusterConfig.getMessageDeserializer(),
-// serviceConfig,
-// new LoadBalancedClusterMessageSender()ErrorHandlerFactory(),
-// clusterConfig.getMessageHandler());
+ return new GrpcSagaClientMessageSender(
+ address,
+ channel,
+ clusterConfig.getMessageSerializer(),
+ clusterConfig.getMessageDeserializer(),
+ serviceConfig,
+ clusterConfig.getMessageHandler(),
+ loadContext
+ );
default:
}
return null;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceSenderAdapter.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceSenderAdapter.java
index d9836d6..5caad87 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceSenderAdapter.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceSenderAdapter.java
@@ -101,4 +101,8 @@ public abstract class LoadBalanceSenderAdapter implements MessageSender {
public MessageSenderPicker getSenderPicker() {
return senderPicker;
}
+
+ public LoadBalanceContext getLoadContext() {
+ return loadContext;
+ }
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java
index 3296b1a..2e9161d 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java
@@ -38,16 +38,18 @@ import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.concurrent.Callable;
import javax.net.ssl.SSLException;
-import org.apache.servicecomb.saga.omega.connector.grpc.saga.LoadBalancedClusterMessageSender;
+import org.apache.servicecomb.saga.omega.connector.grpc.saga.SagaLoadBalanceSender;
+import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContext;
+import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContextBuilder;
+import org.apache.servicecomb.saga.omega.connector.grpc.tcc.TransactionType;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
-import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
import org.junit.BeforeClass;
import org.junit.Test;
public class LoadBalanceClusterMessageSenderWithTLSTest extends LoadBalancedClusterMessageSenderTestBase {
@Override
- protected SagaMessageSender newMessageSender(String[] addresses) {
+ protected SagaLoadBalanceSender newMessageSender(String[] addresses) {
ClassLoader classLoader = getClass().getClassLoader();
AlphaClusterConfig clusterConfig = AlphaClusterConfig.builder()
.addresses(ImmutableList.copyOf(addresses))
@@ -61,10 +63,12 @@ public class LoadBalanceClusterMessageSenderWithTLSTest extends LoadBalancedClus
.messageDeserializer(deserializer)
.build();
- return new LoadBalancedClusterMessageSender(
+ LoadBalanceContext loadContext = new LoadBalanceContextBuilder(
+ TransactionType.SAGA,
clusterConfig,
- new ServiceConfig(serviceName),
- 100);
+ new ServiceConfig(serviceName), 100).build();
+
+ return new SagaLoadBalanceSender(loadContext, new FastestSender());
}
@BeforeClass
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
index 0445814..406dcc9 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
@@ -33,8 +33,13 @@ import com.google.common.collect.ImmutableList;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
import java.util.concurrent.Callable;
-import org.apache.servicecomb.saga.omega.connector.grpc.saga.LoadBalancedClusterMessageSender;
+import org.apache.servicecomb.saga.omega.connector.grpc.saga.SagaLoadBalanceSender;
+import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContext;
+import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContextBuilder;
+import org.apache.servicecomb.saga.omega.connector.grpc.tcc.TransactionType;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
@@ -48,7 +53,7 @@ import org.mockito.Mockito;
public class LoadBalancedClusterMessageSenderTest extends LoadBalancedClusterMessageSenderTestBase {
@Override
- protected SagaMessageSender newMessageSender(String[] addresses) {
+ protected SagaLoadBalanceSender newMessageSender(String[] addresses) {
AlphaClusterConfig clusterConfig = AlphaClusterConfig.builder()
.addresses(ImmutableList.copyOf(addresses))
.enableSSL(false)
@@ -57,10 +62,13 @@ public class LoadBalancedClusterMessageSenderTest extends LoadBalancedClusterMes
.messageDeserializer(deserializer)
.messageHandler(handler)
.build();
- return new LoadBalancedClusterMessageSender(
+
+ LoadBalanceContext loadContext = new LoadBalanceContextBuilder(
+ TransactionType.SAGA,
clusterConfig,
- new ServiceConfig(serviceName),
- 100);
+ new ServiceConfig(serviceName), 100).build();
+
+ return new SagaLoadBalanceSender(loadContext, new FastestSender());
}
@BeforeClass
@@ -139,13 +147,12 @@ public class LoadBalancedClusterMessageSenderTest extends LoadBalancedClusterMes
});
}
- @Test(timeout = 1000)
+ @Test
public void stopSendingOnInterruption() throws Exception {
SagaMessageSender underlying = Mockito.mock(SagaMessageSender.class);
doThrow(RuntimeException.class).when(underlying).send(event);
- final SagaMessageSender messageSender = new LoadBalancedClusterMessageSender(underlying);
-
+ setSenders(underlying);
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
@@ -189,9 +196,9 @@ public class LoadBalancedClusterMessageSenderTest extends LoadBalancedClusterMes
SagaMessageSender underlying2 = Mockito.mock(SagaMessageSender.class);
- SagaMessageSender sender = new LoadBalancedClusterMessageSender(underlying1, underlying2);
+ setSenders(underlying1, underlying2);
- sender.onConnected();
+ messageSender.onConnected();
verify(underlying1).onConnected();
verify(underlying2).onConnected();
@@ -204,9 +211,9 @@ public class LoadBalancedClusterMessageSenderTest extends LoadBalancedClusterMes
SagaMessageSender underlying2 = Mockito.mock(SagaMessageSender.class);
- MessageSender sender = new LoadBalancedClusterMessageSender(underlying1, underlying2);
+ setSenders(underlying1, underlying2);
- sender.onDisconnected();
+ messageSender.onDisconnected();
verify(underlying1).onDisconnected();
verify(underlying2).onDisconnected();
@@ -319,4 +326,12 @@ public class LoadBalancedClusterMessageSenderTest extends LoadBalancedClusterMes
}
throw new IllegalStateException("None of the servers received any message");
}
+
+ private void setSenders(SagaMessageSender ... underlyings) {
+ Map<MessageSender, Long> senders = new HashMap<>();
+ for (SagaMessageSender each : underlyings) {
+ senders.put(each, 0L);
+ }
+ messageSender.getLoadContext().setSenders(senders);
+ }
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTestBase.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTestBase.java
index b1d77a4..4e96bcd 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTestBase.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTestBase.java
@@ -28,6 +28,7 @@ import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.apache.servicecomb.saga.common.EventType;
+import org.apache.servicecomb.saga.omega.connector.grpc.saga.SagaLoadBalanceSender;
import org.apache.servicecomb.saga.omega.transaction.MessageDeserializer;
import org.apache.servicecomb.saga.omega.transaction.MessageHandler;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
@@ -107,7 +108,7 @@ public abstract class LoadBalancedClusterMessageSenderTestBase {
protected final String[] addresses = {"localhost:8080", "localhost:8090"};
- protected final SagaMessageSender messageSender = newMessageSender(addresses);
+ protected final SagaLoadBalanceSender messageSender = newMessageSender(addresses);
@AfterClass
public static void tearDown() throws Exception {
@@ -116,7 +117,7 @@ public abstract class LoadBalancedClusterMessageSenderTestBase {
}
}
- protected abstract SagaMessageSender newMessageSender(String[] addresses);
+ protected abstract SagaLoadBalanceSender newMessageSender(String[] addresses);
@After
public void after() throws Exception {
diff --git a/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java b/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
index 49e6ddd..ddaee92 100644
--- a/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
+++ b/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
@@ -20,12 +20,10 @@ package org.apache.servicecomb.saga.omega.spring;
import com.google.common.collect.ImmutableList;
import org.apache.servicecomb.saga.omega.connector.grpc.AlphaClusterConfig;
import org.apache.servicecomb.saga.omega.connector.grpc.FastestSender;
-import org.apache.servicecomb.saga.omega.connector.grpc.saga.LoadBalancedClusterMessageSender;
+import org.apache.servicecomb.saga.omega.connector.grpc.saga.SagaLoadBalanceSender;
import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContext;
import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContextBuilder;
import org.apache.servicecomb.saga.omega.connector.grpc.tcc.TccLoadBalanceSender;
-import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
-import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageSender;
import org.apache.servicecomb.saga.omega.connector.grpc.tcc.TransactionType;
import org.apache.servicecomb.saga.omega.context.CallbackContext;
import org.apache.servicecomb.saga.omega.context.IdGenerator;
@@ -35,9 +33,11 @@ import org.apache.servicecomb.saga.omega.context.UniqueIdGenerator;
import org.apache.servicecomb.saga.omega.format.KryoMessageFormat;
import org.apache.servicecomb.saga.omega.format.MessageFormat;
import org.apache.servicecomb.saga.omega.transaction.MessageHandler;
+import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
import org.apache.servicecomb.saga.omega.transaction.tcc.DefaultParametersContext;
import org.apache.servicecomb.saga.omega.transaction.tcc.ParametersContext;
import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageHandler;
+import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageSender;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
@@ -104,30 +104,57 @@ class OmegaSpringConfig {
return clusterConfig;
}
- @Bean
- SagaMessageSender grpcMessageSender(
+ @Bean(name = "sagaLoadContext")
+ LoadBalanceContext sagaLoadBalanceSenderContext(
AlphaClusterConfig alphaClusterConfig,
ServiceConfig serviceConfig,
@Value("${omega.connection.reconnectDelay:3000}") int reconnectDelay) {
-
- final SagaMessageSender sender = new LoadBalancedClusterMessageSender(
+ LoadBalanceContext loadBalanceSenderContext = new LoadBalanceContextBuilder(
+ TransactionType.SAGA,
alphaClusterConfig,
serviceConfig,
- reconnectDelay);
+ reconnectDelay).build();
+ return loadBalanceSenderContext;
+ }
- sender.onConnected();
-
+// @Bean
+// SagaMessageSender grpcMessageSender(
+// AlphaClusterConfig alphaClusterConfig,
+// ServiceConfig serviceConfig,
+// @Value("${omega.connection.reconnectDelay:3000}") int reconnectDelay) {
+//
+// final SagaMessageSender sender = new LoadBalancedClusterMessageSender(
+// alphaClusterConfig,
+// serviceConfig,
+// reconnectDelay);
+//
+// sender.onConnected();
+//
+// Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
+// @Override
+// public void run() {
+// sender.onDisconnected();
+// sender.close();
+// }
+// }));
+// return sender;
+// }
+
+ @Bean
+ SagaMessageSender sagaLoadBalanceSender(@Qualifier("sagaLoadContext") LoadBalanceContext loadBalanceSenderContext) {
+ final SagaMessageSender sagaMessageSender = new SagaLoadBalanceSender(loadBalanceSenderContext, new FastestSender());
+ sagaMessageSender.onConnected();
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
@Override
public void run() {
- sender.onDisconnected();
- sender.close();
+ sagaMessageSender.onDisconnected();
+ sagaMessageSender.close();
}
}));
- return sender;
+ return sagaMessageSender;
}
- @Bean
+ @Bean(name = "tccLoadContext")
LoadBalanceContext loadBalanceSenderContext(
AlphaClusterConfig alphaClusterConfig,
ServiceConfig serviceConfig,
@@ -141,7 +168,7 @@ class OmegaSpringConfig {
}
@Bean
- TccMessageSender tccLoadBalanceSender(LoadBalanceContext loadBalanceSenderContext) {
+ TccMessageSender tccLoadBalanceSender(@Qualifier("tccLoadContext") LoadBalanceContext loadBalanceSenderContext) {
final TccMessageSender tccMessageSender = new TccLoadBalanceSender(loadBalanceSenderContext, new FastestSender());
tccMessageSender.onConnected();
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
[incubator-servicecomb-saga] 03/06: SCB-909 Adjust saga load
balance sender package.
Posted by ni...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ningjiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-saga.git
commit ddcc55db698a6b9d4571fc77becc6cdf143ebfa5
Author: cherrylzhao <zh...@126.com>
AuthorDate: Fri Sep 21 15:30:35 2018 +0800
SCB-909 Adjust saga load balance sender package.
---
.../grpc/{ => saga}/GrpcClientMessageSender.java | 26 +++++++++-------------
.../{ => saga}/GrpcCompensateStreamObserver.java | 17 ++++++--------
.../LoadBalancedClusterMessageSender.java | 20 ++++++++++-------
.../grpc/{ => saga}/RetryableMessageSender.java | 14 ++++++------
...LoadBalanceClusterMessageSenderWithTLSTest.java | 2 +-
.../grpc/LoadBalancedClusterMessageSenderTest.java | 1 +
.../connector/grpc/RetryableMessageSenderTest.java | 1 +
.../saga/omega/spring/OmegaSpringConfig.java | 2 +-
8 files changed, 40 insertions(+), 43 deletions(-)
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcClientMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcClientMessageSender.java
similarity index 87%
rename from omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcClientMessageSender.java
rename to omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcClientMessageSender.java
index 234eb95..86a59d7 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcClientMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcClientMessageSender.java
@@ -1,5 +1,4 @@
/*
- *
* 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.
@@ -7,25 +6,24 @@
* (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.
- *
+ * 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.servicecomb.saga.omega.connector.grpc;
+package org.apache.servicecomb.saga.omega.connector.grpc.saga;
-import org.apache.servicecomb.saga.omega.connector.grpc.LoadBalancedClusterMessageSender.ErrorHandlerFactory;
+import com.google.protobuf.ByteString;
+import io.grpc.ManagedChannel;
+import org.apache.servicecomb.saga.omega.connector.grpc.saga.LoadBalancedClusterMessageSender.ErrorHandlerFactory;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.MessageDeserializer;
import org.apache.servicecomb.saga.omega.transaction.MessageHandler;
-import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.MessageSerializer;
import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
import org.apache.servicecomb.saga.omega.transaction.TxEvent;
@@ -37,10 +35,6 @@ import org.apache.servicecomb.saga.pack.contract.grpc.TxEventServiceGrpc;
import org.apache.servicecomb.saga.pack.contract.grpc.TxEventServiceGrpc.TxEventServiceBlockingStub;
import org.apache.servicecomb.saga.pack.contract.grpc.TxEventServiceGrpc.TxEventServiceStub;
-import com.google.protobuf.ByteString;
-
-import io.grpc.ManagedChannel;
-
public class GrpcClientMessageSender implements SagaMessageSender {
private final String target;
private final TxEventServiceStub asyncEventService;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcCompensateStreamObserver.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcCompensateStreamObserver.java
similarity index 82%
rename from omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcCompensateStreamObserver.java
rename to omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcCompensateStreamObserver.java
index 6aae96a..f9a18b1 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcCompensateStreamObserver.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcCompensateStreamObserver.java
@@ -1,5 +1,4 @@
/*
- *
* 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.
@@ -7,18 +6,16 @@
* (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.
- *
+ * 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.servicecomb.saga.omega.connector.grpc;
+package org.apache.servicecomb.saga.omega.connector.grpc.saga;
import java.lang.invoke.MethodHandles;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/LoadBalancedClusterMessageSender.java
similarity index 90%
rename from omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java
rename to omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/LoadBalancedClusterMessageSender.java
index 8620fc9..da02590 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/LoadBalancedClusterMessageSender.java
@@ -6,16 +6,16 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc;
+package org.apache.servicecomb.saga.omega.connector.grpc.saga;
import static java.util.Collections.emptyList;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
@@ -43,6 +43,10 @@ import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import javax.net.ssl.SSLException;
+import org.apache.servicecomb.saga.omega.connector.grpc.AlphaClusterConfig;
+import org.apache.servicecomb.saga.omega.connector.grpc.FastestSender;
+import org.apache.servicecomb.saga.omega.connector.grpc.MessageSenderPicker;
+import org.apache.servicecomb.saga.omega.connector.grpc.PushBackReconnectRunnable;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
@@ -118,7 +122,7 @@ public class LoadBalancedClusterMessageSender implements SagaMessageSender {
}
// this is for test only
- LoadBalancedClusterMessageSender(SagaMessageSender... messageSenders) {
+ public LoadBalancedClusterMessageSender(SagaMessageSender... messageSenders) {
for (SagaMessageSender sender : messageSenders) {
senders.put(sender, 0L);
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/RetryableMessageSender.java
similarity index 80%
rename from omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSender.java
rename to omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/RetryableMessageSender.java
index b052716..b25b9f4 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/RetryableMessageSender.java
@@ -6,16 +6,16 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc;
+package org.apache.servicecomb.saga.omega.connector.grpc.saga;
import static org.apache.servicecomb.saga.common.EventType.SagaStartedEvent;
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java
index 8b241c9..3296b1a 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java
@@ -38,8 +38,8 @@ import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.concurrent.Callable;
import javax.net.ssl.SSLException;
+import org.apache.servicecomb.saga.omega.connector.grpc.saga.LoadBalancedClusterMessageSender;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
-import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
import org.junit.BeforeClass;
import org.junit.Test;
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
index bf413b2..0445814 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
@@ -34,6 +34,7 @@ import io.grpc.Server;
import io.grpc.ServerBuilder;
import java.io.IOException;
import java.util.concurrent.Callable;
+import org.apache.servicecomb.saga.omega.connector.grpc.saga.LoadBalancedClusterMessageSender;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSenderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSenderTest.java
index 2446d31..9f97eb6 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSenderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSenderTest.java
@@ -27,6 +27,7 @@ import static org.mockito.Mockito.verify;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
+import org.apache.servicecomb.saga.omega.connector.grpc.saga.RetryableMessageSender;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
diff --git a/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java b/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
index 054208d..49e6ddd 100644
--- a/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
+++ b/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
@@ -20,7 +20,7 @@ package org.apache.servicecomb.saga.omega.spring;
import com.google.common.collect.ImmutableList;
import org.apache.servicecomb.saga.omega.connector.grpc.AlphaClusterConfig;
import org.apache.servicecomb.saga.omega.connector.grpc.FastestSender;
-import org.apache.servicecomb.saga.omega.connector.grpc.LoadBalancedClusterMessageSender;
+import org.apache.servicecomb.saga.omega.connector.grpc.saga.LoadBalancedClusterMessageSender;
import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContext;
import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContextBuilder;
import org.apache.servicecomb.saga.omega.connector.grpc.tcc.TccLoadBalanceSender;
[incubator-servicecomb-saga] 06/06: SCB-909 Revise copyright space.
Posted by ni...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ningjiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-saga.git
commit 5350a86c0e3137db50db305d2788de2b3f357960
Author: cherrylzhao <zh...@126.com>
AuthorDate: Fri Sep 21 19:09:48 2018 +0800
SCB-909 Revise copyright space.
---
.../saga/omega/connector/grpc/core/FastestSender.java | 12 ++++++------
.../saga/omega/connector/grpc/core/GrpcOnErrorHandler.java | 12 ++++++------
.../saga/omega/connector/grpc/core/LoadBalanceContext.java | 12 ++++++------
.../omega/connector/grpc/core/LoadBalanceContextBuilder.java | 12 ++++++------
.../omega/connector/grpc/core/LoadBalanceSenderAdapter.java | 12 ++++++------
.../saga/omega/connector/grpc/core/MessageSenderPicker.java | 12 ++++++------
.../saga/omega/connector/grpc/core/PendingTaskRunner.java | 12 ++++++------
.../omega/connector/grpc/core/PushBackReconnectRunnable.java | 12 ++++++------
.../omega/connector/grpc/core/ReconnectStreamObserver.java | 12 ++++++------
.../saga/omega/connector/grpc/core/SenderExecutor.java | 12 ++++++------
.../saga/omega/connector/grpc/core/TransactionType.java | 12 ++++++------
.../connector/grpc/saga/GrpcCompensateStreamObserver.java | 12 ++++++------
.../connector/grpc/saga/GrpcSagaClientMessageSender.java | 12 ++++++------
.../omega/connector/grpc/saga/RetryableMessageSender.java | 12 ++++++------
.../omega/connector/grpc/saga/SagaLoadBalanceSender.java | 12 ++++++------
.../connector/grpc/tcc/GrpcCoordinateStreamObserver.java | 12 ++++++------
.../omega/connector/grpc/tcc/GrpcTccClientMessageSender.java | 12 ++++++------
.../saga/omega/connector/grpc/tcc/TccLoadBalanceSender.java | 12 ++++++------
.../connector/grpc/tcc/LoadBalanceContextBuilderTest.java | 12 ++++++------
.../omega/connector/grpc/tcc/LoadBalanceSenderTestBase.java | 12 ++++++------
.../saga/omega/connector/grpc/tcc/MyTccEventServiceImpl.java | 12 ++++++------
.../omega/connector/grpc/tcc/SagaLoadBalanceSenderTest.java | 12 ++++++------
.../omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java | 12 ++++++------
23 files changed, 138 insertions(+), 138 deletions(-)
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/FastestSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/FastestSender.java
index 8c4fc99..486c1b7 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/FastestSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/FastestSender.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.core;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/GrpcOnErrorHandler.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/GrpcOnErrorHandler.java
index 0e7432e..7ab02a1 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/GrpcOnErrorHandler.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/GrpcOnErrorHandler.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.core;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceContext.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceContext.java
index b951a18..2918be8 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceContext.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceContext.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.core;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceContextBuilder.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceContextBuilder.java
index 5bd7205..3f3d542 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceContextBuilder.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceContextBuilder.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.core;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceSenderAdapter.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceSenderAdapter.java
index e06efba..424dbbe 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceSenderAdapter.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceSenderAdapter.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.core;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/MessageSenderPicker.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/MessageSenderPicker.java
index 6bb92d7..c4b669d 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/MessageSenderPicker.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/MessageSenderPicker.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.core;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/PendingTaskRunner.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/PendingTaskRunner.java
index 7e1cbf2..9bb7c95 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/PendingTaskRunner.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/PendingTaskRunner.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.core;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/PushBackReconnectRunnable.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/PushBackReconnectRunnable.java
index 9351f51..bc3c1b3 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/PushBackReconnectRunnable.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/PushBackReconnectRunnable.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.core;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/ReconnectStreamObserver.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/ReconnectStreamObserver.java
index 809a091..bc6971d 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/ReconnectStreamObserver.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/ReconnectStreamObserver.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.core;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/SenderExecutor.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/SenderExecutor.java
index 5958150..ed0a03f 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/SenderExecutor.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/SenderExecutor.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.core;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/TransactionType.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/TransactionType.java
index 36b776c..236bb53 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/TransactionType.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/TransactionType.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.core;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcCompensateStreamObserver.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcCompensateStreamObserver.java
index 8109c00..cdaec10 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcCompensateStreamObserver.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcCompensateStreamObserver.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.saga;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcSagaClientMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcSagaClientMessageSender.java
index caa9c82..5770920 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcSagaClientMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcSagaClientMessageSender.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.saga;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/RetryableMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/RetryableMessageSender.java
index b25b9f4..6869dae 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/RetryableMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/RetryableMessageSender.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.saga;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/SagaLoadBalanceSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/SagaLoadBalanceSender.java
index 7cdcdc6..df2a367 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/SagaLoadBalanceSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/SagaLoadBalanceSender.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.saga;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcCoordinateStreamObserver.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcCoordinateStreamObserver.java
index d334fee..e2aec97 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcCoordinateStreamObserver.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcCoordinateStreamObserver.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.tcc;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java
index 48c364b..c0020f0 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.tcc;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSender.java
index 9608163..a1e762c 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSender.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.tcc;
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilderTest.java
index 23d3d4d..6be785b 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilderTest.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.tcc;
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceSenderTestBase.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceSenderTestBase.java
index b95e9bc..a2691a7 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceSenderTestBase.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceSenderTestBase.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.tcc;
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/MyTccEventServiceImpl.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/MyTccEventServiceImpl.java
index bebe7b2..f2e10cb 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/MyTccEventServiceImpl.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/MyTccEventServiceImpl.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.tcc;
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/SagaLoadBalanceSenderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/SagaLoadBalanceSenderTest.java
index cd8eab7..3adeee7 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/SagaLoadBalanceSenderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/SagaLoadBalanceSenderTest.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.tcc;
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java
index 9f1f0de..ea7cebf 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java
@@ -6,13 +6,13 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc.tcc;
[incubator-servicecomb-saga] 05/06: SCB-909 Optimize package of
load balance.
Posted by ni...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ningjiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-saga.git
commit cd148fd1792df79e28aa6a3847017cada375cbaa
Author: cherrylzhao <zh...@126.com>
AuthorDate: Fri Sep 21 17:23:19 2018 +0800
SCB-909 Optimize package of load balance.
---
.../connector/grpc/{ => core}/FastestSender.java | 2 +-
.../grpc/{tcc => core}/GrpcOnErrorHandler.java | 3 +-
.../grpc/{tcc => core}/LoadBalanceContext.java | 2 +-
.../{tcc => core}/LoadBalanceContextBuilder.java | 3 +-
.../{tcc => core}/LoadBalanceSenderAdapter.java | 3 +-
.../grpc/{ => core}/MessageSenderPicker.java | 14 +-
.../grpc/{tcc => core}/PendingTaskRunner.java | 2 +-
.../grpc/{ => core}/PushBackReconnectRunnable.java | 14 +-
.../{tcc => core}/ReconnectStreamObserver.java | 3 +-
.../grpc/{tcc => core}/SenderExecutor.java | 2 +-
.../grpc/{tcc => core}/TransactionType.java | 2 +-
.../grpc/saga/GrpcCompensateStreamObserver.java | 6 +-
.../grpc/saga/GrpcSagaClientMessageSender.java | 2 +-
.../saga/LoadBalancedClusterMessageSender.java | 246 ---------------------
.../connector/grpc/saga/SagaLoadBalanceSender.java | 8 +-
.../grpc/tcc/GrpcCoordinateStreamObserver.java | 2 +
.../grpc/tcc/GrpcTccClientMessageSender.java | 1 +
.../connector/grpc/tcc/TccLoadBalanceSender.java | 5 +-
...LoadBalanceClusterMessageSenderWithTLSTest.java | 7 +-
.../grpc/LoadBalancedClusterMessageSenderTest.java | 7 +-
.../grpc/PushBackReconnectRunnableTest.java | 1 +
.../grpc/tcc/LoadBalanceContextBuilderTest.java | 27 ++-
.../grpc/tcc/TccLoadBalanceSenderTest.java | 5 +-
.../saga/omega/spring/OmegaSpringConfig.java | 31 +--
24 files changed, 80 insertions(+), 318 deletions(-)
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/FastestSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/FastestSender.java
similarity index 96%
rename from omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/FastestSender.java
rename to omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/FastestSender.java
index 206a447..8c4fc99 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/FastestSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/FastestSender.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.servicecomb.saga.omega.connector.grpc;
+package org.apache.servicecomb.saga.omega.connector.grpc.core;
import com.google.common.base.Supplier;
import java.util.Map;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcOnErrorHandler.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/GrpcOnErrorHandler.java
similarity index 94%
rename from omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcOnErrorHandler.java
rename to omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/GrpcOnErrorHandler.java
index 159ca1f..0e7432e 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcOnErrorHandler.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/GrpcOnErrorHandler.java
@@ -15,13 +15,12 @@
* limitations under the License.
*/
-package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
+package org.apache.servicecomb.saga.omega.connector.grpc.core;
import com.google.common.base.Supplier;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
-import org.apache.servicecomb.saga.omega.connector.grpc.PushBackReconnectRunnable;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContext.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceContext.java
similarity index 97%
rename from omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContext.java
rename to omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceContext.java
index 86c1991..b951a18 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContext.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceContext.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
+package org.apache.servicecomb.saga.omega.connector.grpc.core;
import io.grpc.ManagedChannel;
import java.util.Collection;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilder.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceContextBuilder.java
similarity index 97%
rename from omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilder.java
rename to omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceContextBuilder.java
index d4411ec..5bd7205 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilder.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceContextBuilder.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
+package org.apache.servicecomb.saga.omega.connector.grpc.core;
import com.google.common.base.Optional;
import io.grpc.ManagedChannel;
@@ -37,6 +37,7 @@ import java.util.concurrent.ConcurrentHashMap;
import javax.net.ssl.SSLException;
import org.apache.servicecomb.saga.omega.connector.grpc.AlphaClusterConfig;
import org.apache.servicecomb.saga.omega.connector.grpc.saga.GrpcSagaClientMessageSender;
+import org.apache.servicecomb.saga.omega.connector.grpc.tcc.GrpcTccClientMessageSender;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceSenderAdapter.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceSenderAdapter.java
similarity index 96%
rename from omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceSenderAdapter.java
rename to omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceSenderAdapter.java
index 5caad87..e06efba 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceSenderAdapter.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/LoadBalanceSenderAdapter.java
@@ -15,12 +15,11 @@
* limitations under the License.
*/
-package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
+package org.apache.servicecomb.saga.omega.connector.grpc.core;
import com.google.common.base.Optional;
import io.grpc.ManagedChannel;
import java.lang.invoke.MethodHandles;
-import org.apache.servicecomb.saga.omega.connector.grpc.MessageSenderPicker;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/MessageSenderPicker.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/MessageSenderPicker.java
similarity index 72%
rename from omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/MessageSenderPicker.java
rename to omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/MessageSenderPicker.java
index 3d389b3..6bb92d7 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/MessageSenderPicker.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/MessageSenderPicker.java
@@ -6,16 +6,16 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc;
+package org.apache.servicecomb.saga.omega.connector.grpc.core;
import com.google.common.base.Supplier;
import java.util.Collection;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/PendingTaskRunner.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/PendingTaskRunner.java
similarity index 96%
rename from omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/PendingTaskRunner.java
rename to omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/PendingTaskRunner.java
index dfbfbd8..7e1cbf2 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/PendingTaskRunner.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/PendingTaskRunner.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
+package org.apache.servicecomb.saga.omega.connector.grpc.core;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnable.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/PushBackReconnectRunnable.java
similarity index 84%
rename from omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnable.java
rename to omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/PushBackReconnectRunnable.java
index 6faffcb..9351f51 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnable.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/PushBackReconnectRunnable.java
@@ -6,16 +6,16 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.connector.grpc;
+package org.apache.servicecomb.saga.omega.connector.grpc.core;
import java.lang.invoke.MethodHandles;
import java.util.Map;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/ReconnectStreamObserver.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/ReconnectStreamObserver.java
similarity index 92%
rename from omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/ReconnectStreamObserver.java
rename to omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/ReconnectStreamObserver.java
index 938656f..809a091 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/ReconnectStreamObserver.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/ReconnectStreamObserver.java
@@ -15,10 +15,11 @@
* limitations under the License.
*/
-package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
+package org.apache.servicecomb.saga.omega.connector.grpc.core;
import io.grpc.stub.StreamObserver;
import java.lang.invoke.MethodHandles;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.LoadBalanceContext;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/SenderExecutor.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/SenderExecutor.java
similarity index 93%
rename from omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/SenderExecutor.java
rename to omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/SenderExecutor.java
index fb31eb5..5958150 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/SenderExecutor.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/SenderExecutor.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
+package org.apache.servicecomb.saga.omega.connector.grpc.core;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TransactionType.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/TransactionType.java
similarity index 93%
rename from omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TransactionType.java
rename to omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/TransactionType.java
index 06108dd..36b776c 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TransactionType.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/core/TransactionType.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
+package org.apache.servicecomb.saga.omega.connector.grpc.core;
public enum TransactionType {
TCC, SAGA
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcCompensateStreamObserver.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcCompensateStreamObserver.java
index 73bf0ff..8109c00 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcCompensateStreamObserver.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcCompensateStreamObserver.java
@@ -19,8 +19,8 @@ package org.apache.servicecomb.saga.omega.connector.grpc.saga;
import java.lang.invoke.MethodHandles;
-import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContext;
-import org.apache.servicecomb.saga.omega.connector.grpc.tcc.ReconnectStreamObserver;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.LoadBalanceContext;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.ReconnectStreamObserver;
import org.apache.servicecomb.saga.omega.transaction.MessageDeserializer;
import org.apache.servicecomb.saga.omega.transaction.MessageHandler;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
@@ -28,8 +28,6 @@ import org.apache.servicecomb.saga.pack.contract.grpc.GrpcCompensateCommand;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import io.grpc.stub.StreamObserver;
-
class GrpcCompensateStreamObserver extends ReconnectStreamObserver<GrpcCompensateCommand> {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcSagaClientMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcSagaClientMessageSender.java
index 61be7f4..caa9c82 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcSagaClientMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/GrpcSagaClientMessageSender.java
@@ -19,7 +19,7 @@ package org.apache.servicecomb.saga.omega.connector.grpc.saga;
import com.google.protobuf.ByteString;
import io.grpc.ManagedChannel;
-import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContext;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.LoadBalanceContext;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.MessageDeserializer;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/LoadBalancedClusterMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/LoadBalancedClusterMessageSender.java
deleted file mode 100644
index 849125b..0000000
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/LoadBalancedClusterMessageSender.java
+++ /dev/null
@@ -1,246 +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.servicecomb.saga.omega.connector.grpc.saga;
-
-import static java.util.Collections.emptyList;
-import static java.util.concurrent.TimeUnit.MILLISECONDS;
-
-import com.google.common.base.Supplier;
-import io.grpc.ManagedChannel;
-import io.grpc.ManagedChannelBuilder;
-import io.grpc.netty.GrpcSslContexts;
-import io.grpc.netty.NegotiationType;
-import io.grpc.netty.NettyChannelBuilder;
-import io.netty.handler.ssl.SslContext;
-import io.netty.handler.ssl.SslContextBuilder;
-import io.netty.handler.ssl.SslProvider;
-import java.io.File;
-import java.io.IOException;
-import java.lang.invoke.MethodHandles;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ScheduledExecutorService;
-import javax.net.ssl.SSLException;
-import org.apache.servicecomb.saga.omega.connector.grpc.AlphaClusterConfig;
-import org.apache.servicecomb.saga.omega.connector.grpc.FastestSender;
-import org.apache.servicecomb.saga.omega.connector.grpc.MessageSenderPicker;
-import org.apache.servicecomb.saga.omega.connector.grpc.PushBackReconnectRunnable;
-import org.apache.servicecomb.saga.omega.context.ServiceConfig;
-import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
-import org.apache.servicecomb.saga.omega.transaction.MessageSender;
-import org.apache.servicecomb.saga.omega.transaction.OmegaException;
-import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
-import org.apache.servicecomb.saga.omega.transaction.TxEvent;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class LoadBalancedClusterMessageSender implements SagaMessageSender {
-
- private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final Map<MessageSender, Long> senders = new ConcurrentHashMap<>();
- private final Collection<ManagedChannel> channels;
-
- private final BlockingQueue<Runnable> pendingTasks = new LinkedBlockingQueue<>();
- private final BlockingQueue<MessageSender> availableMessageSenders = new LinkedBlockingQueue<>();
- private final MessageSender retryableMessageSender = new RetryableMessageSender(availableMessageSenders);
-
- private final Supplier<MessageSender> defaultMessageSender = new Supplier<MessageSender>() {
- @Override
- public MessageSender get() {
- return retryableMessageSender;
- }
- };
-
- private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
-
- public LoadBalancedClusterMessageSender(AlphaClusterConfig clusterConfig,
- ServiceConfig serviceConfig,
- int reconnectDelay) {
-
- if (clusterConfig.getAddresses().isEmpty()) {
- throw new IllegalArgumentException("No reachable cluster address provided");
- }
-
- channels = new ArrayList<>(clusterConfig.getAddresses().size());
-
- SslContext sslContext = null;
- for (String address : clusterConfig.getAddresses()) {
- ManagedChannel channel;
-
- if (clusterConfig.isEnableSSL()) {
- if (sslContext == null) {
- try {
- sslContext = buildSslContext(clusterConfig);
- } catch (SSLException e) {
- throw new IllegalArgumentException("Unable to build SslContext", e);
- }
- }
- channel = NettyChannelBuilder.forTarget(address)
- .negotiationType(NegotiationType.TLS)
- .sslContext(sslContext)
- .build();
- } else {
- channel = ManagedChannelBuilder.forTarget(address).usePlaintext()
- .build();
- }
- channels.add(channel);
-// senders.put(
-// new GrpcSagaClientMessageSender(
-// address,
-// channel,
-// clusterConfig.getMessageSerializer(),
-// clusterConfig.getMessageDeserializer(),
-// serviceConfig,
-// new ErrorHandlerFactory(),
-// clusterConfig.getMessageHandler()),
-// 0L);
- }
-
- scheduleReconnectTask(reconnectDelay);
- }
-
- // this is for test only
- public LoadBalancedClusterMessageSender(SagaMessageSender... messageSenders) {
- for (SagaMessageSender sender : messageSenders) {
- senders.put(sender, 0L);
- }
- channels = emptyList();
- }
-
- @Override
- public void onConnected() {
- for(MessageSender sender :senders.keySet()){
- try {
- sender.onConnected();
- } catch (Exception e) {
- LOG.error("Failed connecting to alpha at {}", sender.target(), e);
- }
- }
- }
-
- @Override
- public void onDisconnected() {
- for (MessageSender sender :senders.keySet()) {
- try {
- sender.onDisconnected();
- } catch (Exception e) {
- LOG.error("Failed disconnecting from alpha at {}", sender.target(), e);
- }
- }
- }
-
- @Override
- public void close() {
- scheduler.shutdown();
- for(ManagedChannel channel : channels) {
- channel.shutdownNow();
- }
- }
-
- @Override
- public String target() {
- return "UNKNOWN";
- }
-
- @Override
- public AlphaResponse send(TxEvent event) {
- return send(event, new FastestSender());
- }
-
- AlphaResponse send(TxEvent event, MessageSenderPicker messageSenderPicker) {
- do {
- SagaMessageSender messageSender = (SagaMessageSender) messageSenderPicker.pick(senders, defaultMessageSender);
-
- try {
- long startTime = System.nanoTime();
- AlphaResponse response = messageSender.send(event);
- senders.put(messageSender, System.nanoTime() - startTime);
-
- return response;
- } catch (OmegaException e) {
- throw e;
- } catch (Exception e) {
- LOG.error("Retry sending event {} due to failure", event, e);
-
- // very large latency on exception
- senders.put(messageSender, Long.MAX_VALUE);
- }
- } while (!Thread.currentThread().isInterrupted());
-
- throw new OmegaException("Failed to send event " + event + " due to interruption");
- }
-
- private void scheduleReconnectTask(int reconnectDelay) {
- scheduler.scheduleWithFixedDelay(new Runnable() {
- @Override
- public void run() {
- try {
- pendingTasks.take().run();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
- }
- }, 0, reconnectDelay, MILLISECONDS);
- }
-
- class ErrorHandlerFactory {
- Runnable getHandler(SagaMessageSender messageSender) {
- final Runnable runnable = new PushBackReconnectRunnable(
- messageSender, senders, pendingTasks, availableMessageSenders);
- return new Runnable() {
- @Override
- public void run() {
- pendingTasks.offer(runnable);
- }
- };
- }
-
- }
-
- private static SslContext buildSslContext(AlphaClusterConfig clusterConfig) throws SSLException {
- SslContextBuilder builder = GrpcSslContexts.forClient();
- // openssl must be used because some older JDk does not support cipher suites required by http2,
- // and the performance of JDK ssl is pretty low compared to openssl.
- builder.sslProvider(SslProvider.OPENSSL);
-
- Properties prop = new Properties();
- try {
- prop.load(LoadBalancedClusterMessageSender.class.getClassLoader().getResourceAsStream("ssl.properties"));
- } catch (IOException e) {
- throw new IllegalArgumentException("Unable to read ssl.properties.", e);
- }
-
- builder.protocols(prop.getProperty("protocols").split(","));
- builder.ciphers(Arrays.asList(prop.getProperty("ciphers").split(",")));
- builder.trustManager(new File(clusterConfig.getCertChain()));
-
- if (clusterConfig.isEnableMutualAuth()) {
- builder.keyManager(new File(clusterConfig.getCert()), new File(clusterConfig.getKey()));
- }
-
- return builder.build();
- }
-}
-
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/SagaLoadBalanceSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/SagaLoadBalanceSender.java
index 119a2d3..7cdcdc6 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/SagaLoadBalanceSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/saga/SagaLoadBalanceSender.java
@@ -18,10 +18,10 @@
package org.apache.servicecomb.saga.omega.connector.grpc.saga;
import com.google.common.base.Optional;
-import org.apache.servicecomb.saga.omega.connector.grpc.MessageSenderPicker;
-import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContext;
-import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceSenderAdapter;
-import org.apache.servicecomb.saga.omega.connector.grpc.tcc.SenderExecutor;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.MessageSenderPicker;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.LoadBalanceContext;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.LoadBalanceSenderAdapter;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.SenderExecutor;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcCoordinateStreamObserver.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcCoordinateStreamObserver.java
index d69930c..d334fee 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcCoordinateStreamObserver.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcCoordinateStreamObserver.java
@@ -18,6 +18,8 @@
package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
import java.lang.invoke.MethodHandles;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.LoadBalanceContext;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.ReconnectStreamObserver;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageHandler;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccCoordinateCommand;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java
index a30d6ac..48c364b 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java
@@ -18,6 +18,7 @@
package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
import io.grpc.ManagedChannel;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.LoadBalanceContext;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageHandler;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSender.java
index 6c08bdf..9608163 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSender.java
@@ -18,7 +18,10 @@
package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
import com.google.common.base.Optional;
-import org.apache.servicecomb.saga.omega.connector.grpc.MessageSenderPicker;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.LoadBalanceContext;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.LoadBalanceSenderAdapter;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.MessageSenderPicker;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.SenderExecutor;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageSender;
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java
index 2e9161d..c9d0e02 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java
@@ -38,10 +38,11 @@ import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.concurrent.Callable;
import javax.net.ssl.SSLException;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.FastestSender;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.LoadBalanceContext;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.LoadBalanceContextBuilder;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.TransactionType;
import org.apache.servicecomb.saga.omega.connector.grpc.saga.SagaLoadBalanceSender;
-import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContext;
-import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContextBuilder;
-import org.apache.servicecomb.saga.omega.connector.grpc.tcc.TransactionType;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.junit.BeforeClass;
import org.junit.Test;
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
index 406dcc9..1ffdb3b 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
@@ -36,10 +36,11 @@ import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Callable;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.FastestSender;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.LoadBalanceContext;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.LoadBalanceContextBuilder;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.TransactionType;
import org.apache.servicecomb.saga.omega.connector.grpc.saga.SagaLoadBalanceSender;
-import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContext;
-import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContextBuilder;
-import org.apache.servicecomb.saga.omega.connector.grpc.tcc.TransactionType;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnableTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnableTest.java
index d273dc3..aad214f 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnableTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnableTest.java
@@ -30,6 +30,7 @@ import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.PushBackReconnectRunnable;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.junit.Before;
import org.junit.Test;
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilderTest.java
index 98e5c99..23d3d4d 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilderTest.java
@@ -31,7 +31,11 @@ import io.grpc.testing.GrpcCleanupRule;
import java.io.IOException;
import java.util.ArrayList;
import org.apache.servicecomb.saga.omega.connector.grpc.AlphaClusterConfig;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.LoadBalanceContext;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.LoadBalanceContextBuilder;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.TransactionType;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
+import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
import org.apache.servicecomb.saga.omega.transaction.tcc.CoordinateMessageHandler;
import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageHandler;
import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageSender;
@@ -51,7 +55,7 @@ public class LoadBalanceContextBuilderTest {
private final ServiceConfig serviceConfig = new ServiceConfig(serverName);
protected final String[] addresses = {"localhost:8080", "localhost:8090"};
- private LoadBalanceContextBuilder tccLoadBalanceContextBuilder;
+ private LoadBalanceContextBuilder tccLoadBalanceContextBuilder;
private LoadBalanceContextBuilder sagaLoadBalanceContextBuilder;
@Before
@@ -108,12 +112,29 @@ public class LoadBalanceContextBuilderTest {
@Test
public void buildSagaLoadBalanceContextWithoutSsl() {
-
+ LoadBalanceContext loadContext = sagaLoadBalanceContextBuilder.build();
+ assertThat(loadContext.getPendingTaskRunner().getReconnectDelay(), is(30));
+ assertThat(loadContext.getSenders().size(), is(2));
+ assertThat(loadContext.getSenders().keySet().iterator().next(), instanceOf(SagaMessageSender.class));
+ assertThat(loadContext.getSenders().values().iterator().next(), is(0l));
+ assertThat(loadContext.getChannels().size(), is(2));
+ loadContext.getSenders().keySet().iterator().next().close();
+ shutdownChannels(loadContext);
}
@Test
public void buildSagaLoadBalanceContextWithSsl() {
-
+ when(clusterConfig.isEnableSSL()).thenReturn(true);
+ when(clusterConfig.getCert()).thenReturn(getClass().getClassLoader().getResource("client.crt").getFile());
+ when(clusterConfig.getCertChain()).thenReturn(getClass().getClassLoader().getResource("ca.crt").getFile());
+ when(clusterConfig.getKey()).thenReturn(getClass().getClassLoader().getResource("client.pem").getFile());
+ LoadBalanceContext loadContext = sagaLoadBalanceContextBuilder.build();
+ assertThat(loadContext.getPendingTaskRunner().getReconnectDelay(), is(30));
+ assertThat(loadContext.getSenders().size(), is(2));
+ assertThat(loadContext.getSenders().keySet().iterator().next(), instanceOf(SagaMessageSender.class));
+ assertThat(loadContext.getSenders().values().iterator().next(), is(0l));
+ assertThat(loadContext.getChannels().size(), is(2));
+ shutdownChannels(loadContext);
}
private void shutdownChannels(LoadBalanceContext loadContext) {
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java
index dc5bd11..9f1f0de 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java
@@ -43,7 +43,10 @@ import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import org.apache.servicecomb.saga.common.TransactionStatus;
import org.apache.servicecomb.saga.omega.connector.grpc.AlphaClusterConfig;
-import org.apache.servicecomb.saga.omega.connector.grpc.FastestSender;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.FastestSender;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.LoadBalanceContext;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.LoadBalanceContextBuilder;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.TransactionType;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
diff --git a/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java b/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
index ddaee92..b1fe7e9 100644
--- a/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
+++ b/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
@@ -19,12 +19,12 @@ package org.apache.servicecomb.saga.omega.spring;
import com.google.common.collect.ImmutableList;
import org.apache.servicecomb.saga.omega.connector.grpc.AlphaClusterConfig;
-import org.apache.servicecomb.saga.omega.connector.grpc.FastestSender;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.FastestSender;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.LoadBalanceContext;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.LoadBalanceContextBuilder;
+import org.apache.servicecomb.saga.omega.connector.grpc.core.TransactionType;
import org.apache.servicecomb.saga.omega.connector.grpc.saga.SagaLoadBalanceSender;
-import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContext;
-import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContextBuilder;
import org.apache.servicecomb.saga.omega.connector.grpc.tcc.TccLoadBalanceSender;
-import org.apache.servicecomb.saga.omega.connector.grpc.tcc.TransactionType;
import org.apache.servicecomb.saga.omega.context.CallbackContext;
import org.apache.servicecomb.saga.omega.context.IdGenerator;
import org.apache.servicecomb.saga.omega.context.OmegaContext;
@@ -117,29 +117,6 @@ class OmegaSpringConfig {
return loadBalanceSenderContext;
}
-// @Bean
-// SagaMessageSender grpcMessageSender(
-// AlphaClusterConfig alphaClusterConfig,
-// ServiceConfig serviceConfig,
-// @Value("${omega.connection.reconnectDelay:3000}") int reconnectDelay) {
-//
-// final SagaMessageSender sender = new LoadBalancedClusterMessageSender(
-// alphaClusterConfig,
-// serviceConfig,
-// reconnectDelay);
-//
-// sender.onConnected();
-//
-// Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
-// @Override
-// public void run() {
-// sender.onDisconnected();
-// sender.close();
-// }
-// }));
-// return sender;
-// }
-
@Bean
SagaMessageSender sagaLoadBalanceSender(@Qualifier("sagaLoadContext") LoadBalanceContext loadBalanceSenderContext) {
final SagaMessageSender sagaMessageSender = new SagaLoadBalanceSender(loadBalanceSenderContext, new FastestSender());
[incubator-servicecomb-saga] 02/06: SCB-909 clean up
GrpcTccEventService.
Posted by ni...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ningjiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-saga.git
commit 6594776d37f0d346541a357bc51a7ca59d789dd2
Author: cherrylzhao <zh...@126.com>
AuthorDate: Fri Sep 21 15:23:09 2018 +0800
SCB-909 clean up GrpcTccEventService.
---
.../omega/connector/grpc/GrpcTccEventService.java | 157 ---------------------
...st.java => GrpcTccClientMessageSenderTest.java} | 7 +-
.../saga/omega/spring/OmegaSpringConfig.java | 25 ----
3 files changed, 4 insertions(+), 185 deletions(-)
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcTccEventService.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcTccEventService.java
deleted file mode 100644
index c619584..0000000
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcTccEventService.java
+++ /dev/null
@@ -1,157 +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.servicecomb.saga.omega.connector.grpc;
-
-import org.apache.servicecomb.saga.omega.connector.grpc.tcc.GrpcCoordinateStreamObserver;
-import org.apache.servicecomb.saga.omega.context.ServiceConfig;
-import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
-import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageHandler;
-import org.apache.servicecomb.saga.omega.transaction.tcc.TccEventService;
-import org.apache.servicecomb.saga.omega.transaction.tcc.events.CoordinatedEvent;
-import org.apache.servicecomb.saga.omega.transaction.tcc.events.ParticipatedEvent;
-import org.apache.servicecomb.saga.omega.transaction.tcc.events.TccEndedEvent;
-import org.apache.servicecomb.saga.omega.transaction.tcc.events.TccStartedEvent;
-
-import org.apache.servicecomb.saga.pack.contract.grpc.GrpcAck;
-import org.apache.servicecomb.saga.pack.contract.grpc.GrpcServiceConfig;
-import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccCoordinatedEvent;
-import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccParticipatedEvent;
-import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccTransactionEndedEvent;
-import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccTransactionStartedEvent;
-import org.apache.servicecomb.saga.pack.contract.grpc.TccEventServiceGrpc;
-import org.apache.servicecomb.saga.pack.contract.grpc.TccEventServiceGrpc.TccEventServiceBlockingStub;
-import org.apache.servicecomb.saga.pack.contract.grpc.TccEventServiceGrpc.TccEventServiceStub;
-
-import io.grpc.ManagedChannel;
-
-public class GrpcTccEventService implements TccEventService {
- private final GrpcServiceConfig serviceConfig;
- private final String target;
- private final TccEventServiceBlockingStub tccBlockingEventService;
- private final TccEventServiceStub tccAsyncEventService;
- private final GrpcCoordinateStreamObserver observer;
-
- public GrpcTccEventService(ServiceConfig serviceConfig,
- ManagedChannel channel,
- String address,
- TccMessageHandler handler
- ) {
- this.target = address;
- tccBlockingEventService = TccEventServiceGrpc.newBlockingStub(channel);
- tccAsyncEventService = TccEventServiceGrpc.newStub(channel);
- this.serviceConfig = serviceConfig(serviceConfig.serviceName(), serviceConfig.instanceId());
- observer = new GrpcCoordinateStreamObserver(null, null, handler);
- }
-
- @Override
- public void onConnected() {
- tccAsyncEventService.onConnected(serviceConfig, observer);
- }
-
- @Override
- public void onDisconnected() {
- tccBlockingEventService.onDisconnected(serviceConfig);
- }
-
- @Override
- public void close() {
- // do nothing here
- }
-
- @Override
- public String target() {
- return target;
- }
-
- @Override
- public AlphaResponse participate(ParticipatedEvent participateEvent) {
- GrpcAck grpcAck = tccBlockingEventService.participate(convertTo(participateEvent));
- return new AlphaResponse(grpcAck.getAborted());
- }
-
- @Override
- public AlphaResponse tccTransactionStart(TccStartedEvent tccStartEvent) {
- GrpcAck grpcAck = tccBlockingEventService.onTccTransactionStarted(convertTo(tccStartEvent));
- return new AlphaResponse(grpcAck.getAborted());
- }
-
-
- @Override
- public AlphaResponse tccTransactionStop(TccEndedEvent tccEndEvent) {
- GrpcAck grpcAck = tccBlockingEventService.onTccTransactionEnded(convertTo(tccEndEvent));
- return new AlphaResponse(grpcAck.getAborted());
-
- }
-
- @Override
- public AlphaResponse coordinate(CoordinatedEvent coordinatedEvent) {
- GrpcAck grpcAck = tccBlockingEventService.onTccCoordinated(convertTo(coordinatedEvent));
- return new AlphaResponse(grpcAck.getAborted());
- }
-
- private GrpcTccCoordinatedEvent convertTo(CoordinatedEvent coordinatedEvent) {
- return GrpcTccCoordinatedEvent.newBuilder()
- .setServiceName(serviceConfig.getServiceName())
- .setInstanceId(serviceConfig.getInstanceId())
- .setGlobalTxId(coordinatedEvent.getGlobalTxId())
- .setLocalTxId(coordinatedEvent.getLocalTxId())
- .setParentTxId(coordinatedEvent.getParentTxId())
- .setMethodName(coordinatedEvent.getMethodName())
- .setStatus(coordinatedEvent.getStatus().toString())
- .build();
- }
-
- private GrpcServiceConfig serviceConfig(String serviceName, String instanceId) {
- return GrpcServiceConfig.newBuilder()
- .setServiceName(serviceName)
- .setInstanceId(instanceId)
- .build();
- }
-
- private GrpcTccTransactionStartedEvent convertTo(TccStartedEvent tccStartEvent) {
- return GrpcTccTransactionStartedEvent.newBuilder()
- .setServiceName(serviceConfig.getServiceName())
- .setInstanceId(serviceConfig.getInstanceId())
- .setGlobalTxId(tccStartEvent.getGlobalTxId())
- .setLocalTxId(tccStartEvent.getLocalTxId())
- .build();
- }
-
- private GrpcTccTransactionEndedEvent convertTo(TccEndedEvent tccEndEvent) {
- return GrpcTccTransactionEndedEvent.newBuilder()
- .setServiceName(serviceConfig.getServiceName())
- .setInstanceId(serviceConfig.getInstanceId())
- .setGlobalTxId(tccEndEvent.getGlobalTxId())
- .setLocalTxId(tccEndEvent.getLocalTxId())
- .setStatus(tccEndEvent.getStatus().toString())
- .build();
- }
-
- private GrpcTccParticipatedEvent convertTo(ParticipatedEvent participateEvent) {
- return GrpcTccParticipatedEvent.newBuilder()
- .setServiceName(serviceConfig.getServiceName())
- .setInstanceId(serviceConfig.getInstanceId())
- .setGlobalTxId(participateEvent.getGlobalTxId())
- .setLocalTxId(participateEvent.getLocalTxId())
- .setParentTxId(participateEvent.getParentTxId())
- .setCancelMethod(participateEvent.getCancelMethod())
- .setConfirmMethod(participateEvent.getConfirmMethod())
- .setStatus(participateEvent.getStatus().toString())
- .build();
- }
-}
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcTccEventServiceTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcTccClientMessageSenderTest.java
similarity index 97%
rename from omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcTccEventServiceTest.java
rename to omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcTccClientMessageSenderTest.java
index 51e17af..931e275 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcTccEventServiceTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcTccClientMessageSenderTest.java
@@ -24,6 +24,7 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import org.apache.servicecomb.saga.common.TransactionStatus;
+import org.apache.servicecomb.saga.omega.connector.grpc.tcc.GrpcTccClientMessageSender;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageHandler;
@@ -53,7 +54,7 @@ import io.grpc.testing.GrpcCleanupRule;
import io.grpc.util.MutableHandlerRegistry;
@RunWith(JUnit4.class)
-public class GrpcTccEventServiceTest {
+public class GrpcTccClientMessageSenderTest {
@Rule
public final GrpcCleanupRule grpcCleanup = new GrpcCleanupRule();
private final MutableHandlerRegistry serviceRegistry = new MutableHandlerRegistry();
@@ -70,7 +71,7 @@ public class GrpcTccEventServiceTest {
private final ServiceConfig serviceConfig = new ServiceConfig(uniquify("Service"));
private final String address = uniquify("Address");
private final TccMessageHandler handler = mock(TccMessageHandler.class);
- private GrpcTccEventService service;
+ private GrpcTccClientMessageSender service;
@Before
public void setUp() throws Exception {
@@ -86,7 +87,7 @@ public class GrpcTccEventServiceTest {
InProcessChannelBuilder.forName(serverName).directExecutor().build());
// Create a TccEventServiceStub using the in-process channel;
- service = new GrpcTccEventService(serviceConfig, channel, address, handler);
+ service = new GrpcTccClientMessageSender(serviceConfig, channel, address, handler, null);
}
@Test
diff --git a/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java b/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
index ee887da..054208d 100644
--- a/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
+++ b/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
@@ -18,11 +18,8 @@
package org.apache.servicecomb.saga.omega.spring;
import com.google.common.collect.ImmutableList;
-import io.grpc.ManagedChannel;
-import io.grpc.ManagedChannelBuilder;
import org.apache.servicecomb.saga.omega.connector.grpc.AlphaClusterConfig;
import org.apache.servicecomb.saga.omega.connector.grpc.FastestSender;
-import org.apache.servicecomb.saga.omega.connector.grpc.GrpcTccEventService;
import org.apache.servicecomb.saga.omega.connector.grpc.LoadBalancedClusterMessageSender;
import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContext;
import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContextBuilder;
@@ -38,10 +35,8 @@ import org.apache.servicecomb.saga.omega.context.UniqueIdGenerator;
import org.apache.servicecomb.saga.omega.format.KryoMessageFormat;
import org.apache.servicecomb.saga.omega.format.MessageFormat;
import org.apache.servicecomb.saga.omega.transaction.MessageHandler;
-import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.tcc.DefaultParametersContext;
import org.apache.servicecomb.saga.omega.transaction.tcc.ParametersContext;
-import org.apache.servicecomb.saga.omega.transaction.tcc.TccEventService;
import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageHandler;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
@@ -158,24 +153,4 @@ class OmegaSpringConfig {
}));
return tccMessageSender;
}
-
- // TODO should integrate with loadBalance message sender in future.
- @Bean
- TccEventService tccEventService(ServiceConfig serviceConfig,
- @Lazy TccMessageHandler coordinateMessageHandler,
- @Value("${alpha.cluster.address:localhost:8080}") String address) {
- ManagedChannel channel = ManagedChannelBuilder.forTarget(address).usePlaintext().build();
- final GrpcTccEventService service = new GrpcTccEventService(serviceConfig, channel, address, coordinateMessageHandler);
- // Need to register it self first
- service.onConnected();
-
- Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
- @Override
- public void run() {
- service.onDisconnected();
- service.close();
- }
- }));
- return service;
- }
}
[incubator-servicecomb-saga] 01/06: SCB-909 Integrate
TccLoadBalance to omega-transaction && modify saga message interface.
Posted by ni...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ningjiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-saga.git
commit f4579fed8fe95e3d818e08932e90402f4c92cef6
Author: cherrylzhao <zh...@126.com>
AuthorDate: Fri Sep 21 14:12:25 2018 +0800
SCB-909 Integrate TccLoadBalance to omega-transaction && modify saga message interface.
---
.../omega/connector/grpc/AlphaClusterConfig.java | 1 -
.../saga/omega/connector/grpc/FastestSender.java | 6 ++---
.../connector/grpc/GrpcClientMessageSender.java | 3 ++-
.../grpc/LoadBalancedClusterMessageSender.java | 18 +++++++--------
.../omega/connector/grpc/MessageSenderPicker.java | 2 +-
.../connector/grpc/RetryableMessageSender.java | 6 ++---
.../connector/grpc/tcc/GrpcOnErrorHandler.java | 10 +++++----
.../grpc/tcc/GrpcTccClientMessageSender.java | 7 +-----
.../grpc/tcc/LoadBalanceSenderAdapter.java | 7 ------
.../connector/grpc/tcc/SagaLoadBalanceSender.java | 6 ++---
.../connector/grpc/tcc/TccLoadBalanceSender.java | 1 +
...LoadBalanceClusterMessageSenderWithTLSTest.java | 3 ++-
.../grpc/LoadBalancedClusterMessageSenderTest.java | 17 +++++++-------
.../LoadBalancedClusterMessageSenderTestBase.java | 7 +++---
.../grpc/PushBackReconnectRunnableTest.java | 1 -
.../connector/grpc/RetryableMessageSenderTest.java | 6 ++---
.../grpc/tcc/LoadBalanceContextBuilderTest.java | 1 +
.../grpc/tcc/TccLoadBalanceSenderTest.java | 12 ++++++++--
.../saga/omega/spring/OmegaSpringConfig.java | 23 ++++++++++++++-----
.../spring/TransactionAspectConfig.java | 26 +++++++++++++---------
.../omega/transaction/spring/MessageConfig.java | 10 +++++----
.../omega/transaction/CompensableInterceptor.java | 4 ++--
.../transaction/CompensationMessageHandler.java | 4 ++--
.../saga/omega/transaction/MessageSender.java | 14 ++++++------
.../{MessageSender.java => SagaMessageSender.java} | 21 ++++++-----------
.../transaction/SagaStartAnnotationProcessor.java | 4 ++--
.../saga/omega/transaction/SagaStartAspect.java | 2 +-
.../saga/omega/transaction/TransactionAspect.java | 2 +-
.../transaction/tcc/CoordinateMessageHandler.java | 8 +++----
.../omega/transaction}/tcc/TccMessageSender.java | 2 +-
.../transaction/tcc/TccParticipatorAspect.java | 10 ++++-----
.../tcc/TccStartAnnotationProcessor.java | 13 +++++------
.../saga/omega/transaction/tcc/TccStartAspect.java | 4 ++--
.../transaction/CompensableInterceptorTest.java | 2 +-
.../CompensationMessageHandlerTest.java | 2 +-
.../omega/transaction/DefaultRecoveryTest.java | 4 ++--
.../omega/transaction/ForwardRecoveryTest.java | 4 ++--
.../SagaStartAnnotationProcessorTest.java | 4 ++--
.../omega/transaction/SagaStartAspectTest.java | 3 +--
.../omega/transaction/TransactionAspectTest.java | 2 +-
.../tcc/CoordinateMessageHandlerTest.java | 6 ++---
.../transaction/tcc/TccParticipatorAspectTest.java | 4 ++--
.../tcc/TccStartAnnotationProcessorTest.java | 4 ++--
.../omega/transaction/tcc/TccStartAspectTest.java | 4 ++--
44 files changed, 152 insertions(+), 148 deletions(-)
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/AlphaClusterConfig.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/AlphaClusterConfig.java
index 99bff3f..43fb8fc 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/AlphaClusterConfig.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/AlphaClusterConfig.java
@@ -19,7 +19,6 @@ package org.apache.servicecomb.saga.omega.connector.grpc;
import java.util.Collections;
import java.util.List;
-import org.apache.servicecomb.saga.omega.connector.grpc.tcc.TccMessageSender;
import org.apache.servicecomb.saga.omega.transaction.MessageDeserializer;
import org.apache.servicecomb.saga.omega.transaction.MessageHandler;
import org.apache.servicecomb.saga.omega.transaction.MessageSerializer;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/FastestSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/FastestSender.java
index ed75840..206a447 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/FastestSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/FastestSender.java
@@ -25,13 +25,11 @@ import org.apache.servicecomb.saga.omega.transaction.MessageSender;
* The strategy of picking the fastest {@link MessageSender}
*/
public class FastestSender implements MessageSenderPicker {
-
@Override
- public MessageSender pick(Map<MessageSender, Long> messageSenders,
- Supplier<MessageSender> defaultSender) {
+ public MessageSender pick(Map<? extends MessageSender, Long> messageSenders, Supplier<MessageSender> defaultSender) {
Long min = Long.MAX_VALUE;
MessageSender sender = null;
- for (Map.Entry<MessageSender, Long> entry : messageSenders.entrySet()) {
+ for (Map.Entry<? extends MessageSender, Long> entry : messageSenders.entrySet()) {
if (entry.getValue() != Long.MAX_VALUE && min > entry.getValue()) {
min = entry.getValue();
sender = entry.getKey();
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcClientMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcClientMessageSender.java
index 13da6cf..234eb95 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcClientMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcClientMessageSender.java
@@ -27,6 +27,7 @@ import org.apache.servicecomb.saga.omega.transaction.MessageDeserializer;
import org.apache.servicecomb.saga.omega.transaction.MessageHandler;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.MessageSerializer;
+import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
import org.apache.servicecomb.saga.omega.transaction.TxEvent;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcAck;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcServiceConfig;
@@ -40,7 +41,7 @@ import com.google.protobuf.ByteString;
import io.grpc.ManagedChannel;
-public class GrpcClientMessageSender implements MessageSender {
+public class GrpcClientMessageSender implements SagaMessageSender {
private final String target;
private final TxEventServiceStub asyncEventService;
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java
index c350755..8620fc9 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSender.java
@@ -47,11 +47,12 @@ import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
+import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
import org.apache.servicecomb.saga.omega.transaction.TxEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class LoadBalancedClusterMessageSender implements MessageSender {
+public class LoadBalancedClusterMessageSender implements SagaMessageSender {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private final Map<MessageSender, Long> senders = new ConcurrentHashMap<>();
@@ -59,8 +60,7 @@ public class LoadBalancedClusterMessageSender implements MessageSender {
private final BlockingQueue<Runnable> pendingTasks = new LinkedBlockingQueue<>();
private final BlockingQueue<MessageSender> availableMessageSenders = new LinkedBlockingQueue<>();
- private final MessageSender retryableMessageSender = new RetryableMessageSender(
- availableMessageSenders);
+ private final MessageSender retryableMessageSender = new RetryableMessageSender(availableMessageSenders);
private final Supplier<MessageSender> defaultMessageSender = new Supplier<MessageSender>() {
@Override
@@ -118,8 +118,8 @@ public class LoadBalancedClusterMessageSender implements MessageSender {
}
// this is for test only
- LoadBalancedClusterMessageSender(MessageSender... messageSenders) {
- for (MessageSender sender : messageSenders) {
+ LoadBalancedClusterMessageSender(SagaMessageSender... messageSenders) {
+ for (SagaMessageSender sender : messageSenders) {
senders.put(sender, 0L);
}
channels = emptyList();
@@ -167,7 +167,7 @@ public class LoadBalancedClusterMessageSender implements MessageSender {
AlphaResponse send(TxEvent event, MessageSenderPicker messageSenderPicker) {
do {
- MessageSender messageSender = messageSenderPicker.pick(senders, defaultMessageSender);
+ SagaMessageSender messageSender = (SagaMessageSender) messageSenderPicker.pick(senders, defaultMessageSender);
try {
long startTime = System.nanoTime();
@@ -202,9 +202,9 @@ public class LoadBalancedClusterMessageSender implements MessageSender {
}
class ErrorHandlerFactory {
- Runnable getHandler(MessageSender messageSender) {
- final Runnable runnable = new PushBackReconnectRunnable(messageSender, senders, pendingTasks,
- availableMessageSenders);
+ Runnable getHandler(SagaMessageSender messageSender) {
+ final Runnable runnable = new PushBackReconnectRunnable(
+ messageSender, senders, pendingTasks, availableMessageSenders);
return new Runnable() {
@Override
public void run() {
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/MessageSenderPicker.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/MessageSenderPicker.java
index 25a88f1..3d389b3 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/MessageSenderPicker.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/MessageSenderPicker.java
@@ -36,6 +36,6 @@ public interface MessageSenderPicker {
* @param defaultSender Default sender provider
* @return The specified one.
*/
- MessageSender pick(Map<MessageSender, Long> messageSenders,
+ MessageSender pick(Map<? extends MessageSender, Long> messageSenders,
Supplier<MessageSender> defaultSender);
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSender.java
index 4eb90e1..b052716 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSender.java
@@ -20,13 +20,13 @@ package org.apache.servicecomb.saga.omega.connector.grpc;
import static org.apache.servicecomb.saga.common.EventType.SagaStartedEvent;
import java.util.concurrent.BlockingQueue;
-
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
+import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
import org.apache.servicecomb.saga.omega.transaction.TxEvent;
-public class RetryableMessageSender implements MessageSender {
+public class RetryableMessageSender implements SagaMessageSender {
private final BlockingQueue<MessageSender> availableMessageSenders;
public RetryableMessageSender(BlockingQueue<MessageSender> availableMessageSenders) {
@@ -59,7 +59,7 @@ public class RetryableMessageSender implements MessageSender {
throw new OmegaException("Failed to process subsequent requests because no alpha server is available");
}
try {
- return availableMessageSenders.take().send(event);
+ return ((SagaMessageSender)availableMessageSenders.take()).send(event);
} catch (InterruptedException e) {
throw new OmegaException("Failed to send event " + event + " due to interruption", e);
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcOnErrorHandler.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcOnErrorHandler.java
index 8e9b26f..159ca1f 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcOnErrorHandler.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcOnErrorHandler.java
@@ -22,8 +22,8 @@ import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.servicecomb.saga.omega.connector.grpc.PushBackReconnectRunnable;
-import org.apache.servicecomb.saga.omega.connector.grpc.RetryableMessageSender;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
+import org.apache.servicecomb.saga.omega.transaction.OmegaException;
public class GrpcOnErrorHandler {
@@ -61,12 +61,14 @@ public class GrpcOnErrorHandler {
private final BlockingQueue<MessageSender> reconnectedSenders = new LinkedBlockingQueue<>();
- private final MessageSender retryMessageSender = new RetryableMessageSender(reconnectedSenders);
-
private final Supplier<MessageSender> defaultMessageSender = new Supplier<MessageSender>() {
@Override
public MessageSender get() {
- return retryMessageSender;
+ try {
+ return reconnectedSenders.take();
+ } catch (InterruptedException e) {
+ throw new OmegaException("Failed to get reconnected sender", e);
+ }
}
};
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java
index e23f366..a30d6ac 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java
@@ -20,8 +20,8 @@ package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
import io.grpc.ManagedChannel;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
-import org.apache.servicecomb.saga.omega.transaction.TxEvent;
import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageHandler;
+import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageSender;
import org.apache.servicecomb.saga.omega.transaction.tcc.events.CoordinatedEvent;
import org.apache.servicecomb.saga.omega.transaction.tcc.events.ParticipatedEvent;
import org.apache.servicecomb.saga.omega.transaction.tcc.events.TccEndedEvent;
@@ -152,9 +152,4 @@ public class GrpcTccClientMessageSender implements TccMessageSender {
.setStatus(participateEvent.getStatus().toString())
.build();
}
-
- @Override
- public AlphaResponse send(TxEvent event) {
- return null;
- }
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceSenderAdapter.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceSenderAdapter.java
index b7a8ef2..d9836d6 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceSenderAdapter.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceSenderAdapter.java
@@ -24,7 +24,6 @@ import org.apache.servicecomb.saga.omega.connector.grpc.MessageSenderPicker;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
-import org.apache.servicecomb.saga.omega.transaction.TxEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -60,7 +59,6 @@ public abstract class LoadBalanceSenderAdapter implements MessageSender {
} catch (Exception e) {
LOG.error("Retry sending event {} due to failure", event, e);
loadContext.getSenders().put(messageSender, Long.MAX_VALUE);
-// loadContext.getGrpcOnErrorHandler().handle(messageSender);
}
return Optional.fromNullable(response);
}
@@ -100,11 +98,6 @@ public abstract class LoadBalanceSenderAdapter implements MessageSender {
return "UNKNOWN";
}
- @Override
- public AlphaResponse send(TxEvent event) {
- return null;
- }
-
public MessageSenderPicker getSenderPicker() {
return senderPicker;
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/SagaLoadBalanceSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/SagaLoadBalanceSender.java
index c394791..ded20d7 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/SagaLoadBalanceSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/SagaLoadBalanceSender.java
@@ -20,11 +20,11 @@ package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
import com.google.common.base.Optional;
import org.apache.servicecomb.saga.omega.connector.grpc.MessageSenderPicker;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
-import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
+import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
import org.apache.servicecomb.saga.omega.transaction.TxEvent;
-public class SagaLoadBalanceSender extends LoadBalanceSenderAdapter {
+public class SagaLoadBalanceSender extends LoadBalanceSenderAdapter implements SagaMessageSender {
public SagaLoadBalanceSender(LoadBalanceContext loadContext,
MessageSenderPicker senderPicker) {
@@ -34,7 +34,7 @@ public class SagaLoadBalanceSender extends LoadBalanceSenderAdapter {
@Override
public AlphaResponse send(TxEvent event) {
do {
- final MessageSender messageSender = pickMessageSender();
+ final SagaMessageSender messageSender = pickMessageSender();
Optional<AlphaResponse> response = doGrpcSend(messageSender, event, new SenderExecutor<TxEvent>() {
@Override
public AlphaResponse apply(TxEvent event) {
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSender.java
index fe20337..6c08bdf 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSender.java
@@ -21,6 +21,7 @@ import com.google.common.base.Optional;
import org.apache.servicecomb.saga.omega.connector.grpc.MessageSenderPicker;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
+import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageSender;
import org.apache.servicecomb.saga.omega.transaction.tcc.events.CoordinatedEvent;
import org.apache.servicecomb.saga.omega.transaction.tcc.events.ParticipatedEvent;
import org.apache.servicecomb.saga.omega.transaction.tcc.events.TccEndedEvent;
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java
index 6de7c01..8b241c9 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalanceClusterMessageSenderWithTLSTest.java
@@ -40,13 +40,14 @@ import java.util.concurrent.Callable;
import javax.net.ssl.SSLException;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
+import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
import org.junit.BeforeClass;
import org.junit.Test;
public class LoadBalanceClusterMessageSenderWithTLSTest extends LoadBalancedClusterMessageSenderTestBase {
@Override
- protected MessageSender newMessageSender(String[] addresses) {
+ protected SagaMessageSender newMessageSender(String[] addresses) {
ClassLoader classLoader = getClass().getClassLoader();
AlphaClusterConfig clusterConfig = AlphaClusterConfig.builder()
.addresses(ImmutableList.copyOf(addresses))
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
index 491e15f..bf413b2 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTest.java
@@ -37,6 +37,7 @@ import java.util.concurrent.Callable;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
+import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
import org.apache.servicecomb.saga.omega.transaction.TxAbortedEvent;
import org.apache.servicecomb.saga.omega.transaction.TxEvent;
import org.apache.servicecomb.saga.omega.transaction.TxStartedEvent;
@@ -46,7 +47,7 @@ import org.mockito.Mockito;
public class LoadBalancedClusterMessageSenderTest extends LoadBalancedClusterMessageSenderTestBase {
@Override
- protected MessageSender newMessageSender(String[] addresses) {
+ protected SagaMessageSender newMessageSender(String[] addresses) {
AlphaClusterConfig clusterConfig = AlphaClusterConfig.builder()
.addresses(ImmutableList.copyOf(addresses))
.enableSSL(false)
@@ -139,10 +140,10 @@ public class LoadBalancedClusterMessageSenderTest extends LoadBalancedClusterMes
@Test(timeout = 1000)
public void stopSendingOnInterruption() throws Exception {
- MessageSender underlying = Mockito.mock(MessageSender.class);
+ SagaMessageSender underlying = Mockito.mock(SagaMessageSender.class);
doThrow(RuntimeException.class).when(underlying).send(event);
- final MessageSender messageSender = new LoadBalancedClusterMessageSender(underlying);
+ final SagaMessageSender messageSender = new LoadBalancedClusterMessageSender(underlying);
Thread thread = new Thread(new Runnable() {
@Override
@@ -182,12 +183,12 @@ public class LoadBalancedClusterMessageSenderTest extends LoadBalancedClusterMes
@Test
public void swallowException_UntilAllSendersConnected() throws Exception {
- MessageSender underlying1 = Mockito.mock(MessageSender.class);
+ SagaMessageSender underlying1 = Mockito.mock(SagaMessageSender.class);
doThrow(RuntimeException.class).when(underlying1).onConnected();
- MessageSender underlying2 = Mockito.mock(MessageSender.class);
+ SagaMessageSender underlying2 = Mockito.mock(SagaMessageSender.class);
- MessageSender sender = new LoadBalancedClusterMessageSender(underlying1, underlying2);
+ SagaMessageSender sender = new LoadBalancedClusterMessageSender(underlying1, underlying2);
sender.onConnected();
@@ -197,10 +198,10 @@ public class LoadBalancedClusterMessageSenderTest extends LoadBalancedClusterMes
@Test
public void swallowException_UntilAllSendersDisconnected() throws Exception {
- MessageSender underlying1 = Mockito.mock(MessageSender.class);
+ SagaMessageSender underlying1 = Mockito.mock(SagaMessageSender.class);
doThrow(RuntimeException.class).when(underlying1).onDisconnected();
- MessageSender underlying2 = Mockito.mock(MessageSender.class);
+ SagaMessageSender underlying2 = Mockito.mock(SagaMessageSender.class);
MessageSender sender = new LoadBalancedClusterMessageSender(underlying1, underlying2);
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTestBase.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTestBase.java
index 9a2db23..b1d77a4 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTestBase.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/LoadBalancedClusterMessageSenderTestBase.java
@@ -21,7 +21,6 @@ import static com.seanyinx.github.unit.scaffolding.Randomness.uniquify;
import static org.junit.Assert.fail;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -29,11 +28,11 @@ import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.apache.servicecomb.saga.common.EventType;
-import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.MessageDeserializer;
import org.apache.servicecomb.saga.omega.transaction.MessageHandler;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.MessageSerializer;
+import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
import org.apache.servicecomb.saga.omega.transaction.TxEvent;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcAck;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcCompensateCommand;
@@ -108,7 +107,7 @@ public abstract class LoadBalancedClusterMessageSenderTestBase {
protected final String[] addresses = {"localhost:8080", "localhost:8090"};
- protected final MessageSender messageSender = newMessageSender(addresses);
+ protected final SagaMessageSender messageSender = newMessageSender(addresses);
@AfterClass
public static void tearDown() throws Exception {
@@ -117,7 +116,7 @@ public abstract class LoadBalancedClusterMessageSenderTestBase {
}
}
- protected abstract MessageSender newMessageSender(String[] addresses);
+ protected abstract SagaMessageSender newMessageSender(String[] addresses);
@After
public void after() throws Exception {
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnableTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnableTest.java
index 4a12f1f..d273dc3 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnableTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnableTest.java
@@ -31,7 +31,6 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
-import org.hamcrest.Matcher;
import org.junit.Before;
import org.junit.Test;
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSenderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSenderTest.java
index c1edb26..2446d31 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSenderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/RetryableMessageSenderTest.java
@@ -27,9 +27,9 @@ import static org.mockito.Mockito.verify;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
-
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
+import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
import org.apache.servicecomb.saga.omega.transaction.SagaStartedEvent;
import org.apache.servicecomb.saga.omega.transaction.TxEvent;
import org.apache.servicecomb.saga.omega.transaction.TxStartedEvent;
@@ -38,7 +38,7 @@ import org.junit.Test;
public class RetryableMessageSenderTest {
@SuppressWarnings("unchecked")
private final BlockingQueue<MessageSender> availableMessageSenders = new LinkedBlockingQueue<>();
- private final MessageSender messageSender = new RetryableMessageSender(availableMessageSenders);
+ private final SagaMessageSender messageSender = new RetryableMessageSender(availableMessageSenders);
private final String globalTxId = uniquify("globalTxId");
private final String localTxId = uniquify("localTxId");
@@ -47,7 +47,7 @@ public class RetryableMessageSenderTest {
@Test
public void sendEventWhenSenderIsAvailable() {
- MessageSender sender = mock(MessageSender.class);
+ SagaMessageSender sender = mock(SagaMessageSender.class);
availableMessageSenders.add(sender);
messageSender.send(event);
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilderTest.java
index 37e8aed..98e5c99 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilderTest.java
@@ -34,6 +34,7 @@ import org.apache.servicecomb.saga.omega.connector.grpc.AlphaClusterConfig;
import org.apache.servicecomb.saga.omega.context.ServiceConfig;
import org.apache.servicecomb.saga.omega.transaction.tcc.CoordinateMessageHandler;
import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageHandler;
+import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageSender;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java
index afec043..dc5bd11 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccLoadBalanceSenderTest.java
@@ -35,7 +35,6 @@ import com.google.common.collect.Maps;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.netty.NettyServerBuilder;
-import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Iterator;
import java.util.Map;
@@ -51,6 +50,7 @@ import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.OmegaException;
import org.apache.servicecomb.saga.omega.transaction.tcc.CoordinateMessageHandler;
import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageHandler;
+import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageSender;
import org.apache.servicecomb.saga.omega.transaction.tcc.events.CoordinatedEvent;
import org.apache.servicecomb.saga.omega.transaction.tcc.events.ParticipatedEvent;
import org.apache.servicecomb.saga.omega.transaction.tcc.events.TccEndedEvent;
@@ -58,6 +58,7 @@ import org.apache.servicecomb.saga.omega.transaction.tcc.events.TccStartedEvent;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccParticipatedEvent;
import org.hamcrest.core.Is;
import org.junit.After;
+import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
@@ -88,12 +89,19 @@ public class TccLoadBalanceSenderTest extends LoadBalanceSenderTestBase {
private CoordinatedEvent coordinatedEvent;
@BeforeClass
- public static void startServer() throws IOException {
+ public static void startServer() {
for (Integer each : ports) {
startServerOnPort(each);
}
}
+ @AfterClass
+ public static void shutdownServer() {
+ for(Server server: servers.values()) {
+ server.shutdown();
+ }
+ }
+
private static void startServerOnPort(int port) {
ServerBuilder<?> serverBuilder = NettyServerBuilder.forAddress(
new InetSocketAddress("127.0.0.1", port));
diff --git a/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java b/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
index 614038b..ee887da 100644
--- a/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
+++ b/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/saga/omega/spring/OmegaSpringConfig.java
@@ -27,6 +27,8 @@ import org.apache.servicecomb.saga.omega.connector.grpc.LoadBalancedClusterMessa
import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContext;
import org.apache.servicecomb.saga.omega.connector.grpc.tcc.LoadBalanceContextBuilder;
import org.apache.servicecomb.saga.omega.connector.grpc.tcc.TccLoadBalanceSender;
+import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
+import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageSender;
import org.apache.servicecomb.saga.omega.connector.grpc.tcc.TransactionType;
import org.apache.servicecomb.saga.omega.context.CallbackContext;
import org.apache.servicecomb.saga.omega.context.IdGenerator;
@@ -107,13 +109,13 @@ class OmegaSpringConfig {
return clusterConfig;
}
- @Bean(name = "sagaSender")
- MessageSender grpcMessageSender(
+ @Bean
+ SagaMessageSender grpcMessageSender(
AlphaClusterConfig alphaClusterConfig,
ServiceConfig serviceConfig,
@Value("${omega.connection.reconnectDelay:3000}") int reconnectDelay) {
- final MessageSender sender = new LoadBalancedClusterMessageSender(
+ final SagaMessageSender sender = new LoadBalancedClusterMessageSender(
alphaClusterConfig,
serviceConfig,
reconnectDelay);
@@ -143,9 +145,18 @@ class OmegaSpringConfig {
return loadBalanceSenderContext;
}
- @Bean(name = "tccSender")
- TccLoadBalanceSender tccLoadBalanceSender(LoadBalanceContext loadBalanceSenderContext) {
- return new TccLoadBalanceSender(loadBalanceSenderContext, new FastestSender());
+ @Bean
+ TccMessageSender tccLoadBalanceSender(LoadBalanceContext loadBalanceSenderContext) {
+ final TccMessageSender tccMessageSender = new TccLoadBalanceSender(loadBalanceSenderContext, new FastestSender());
+ tccMessageSender.onConnected();
+ Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
+ @Override
+ public void run() {
+ tccMessageSender.onDisconnected();
+ tccMessageSender.close();
+ }
+ }));
+ return tccMessageSender;
}
// TODO should integrate with loadBalance message sender in future.
diff --git a/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionAspectConfig.java b/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionAspectConfig.java
index 5977b47..e5ab968 100644
--- a/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionAspectConfig.java
+++ b/omega/omega-spring-tx/src/main/java/org/apache/servicecomb/saga/omega/transaction/spring/TransactionAspectConfig.java
@@ -21,13 +21,13 @@ import org.apache.servicecomb.saga.omega.context.CallbackContext;
import org.apache.servicecomb.saga.omega.context.OmegaContext;
import org.apache.servicecomb.saga.omega.transaction.CompensationMessageHandler;
import org.apache.servicecomb.saga.omega.transaction.MessageHandler;
-import org.apache.servicecomb.saga.omega.transaction.MessageSender;
+import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
import org.apache.servicecomb.saga.omega.transaction.SagaStartAspect;
import org.apache.servicecomb.saga.omega.transaction.TransactionAspect;
import org.apache.servicecomb.saga.omega.transaction.tcc.CoordinateMessageHandler;
import org.apache.servicecomb.saga.omega.transaction.tcc.ParametersContext;
-import org.apache.servicecomb.saga.omega.transaction.tcc.TccEventService;
import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageHandler;
+import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageSender;
import org.apache.servicecomb.saga.omega.transaction.tcc.TccParticipatorAspect;
import org.apache.servicecomb.saga.omega.transaction.tcc.TccStartAspect;
import org.springframework.beans.factory.annotation.Qualifier;
@@ -40,18 +40,18 @@ import org.springframework.context.annotation.EnableAspectJAutoProxy;
public class TransactionAspectConfig {
@Bean
- MessageHandler messageHandler(@Qualifier("sagaSender") MessageSender sender,
+ MessageHandler messageHandler(SagaMessageSender sender,
@Qualifier("compensationContext") CallbackContext context, OmegaContext omegaContext) {
return new CompensationMessageHandler(sender, context);
}
@Bean
- SagaStartAspect sagaStartAspect(@Qualifier("sagaSender")MessageSender sender, OmegaContext context) {
+ SagaStartAspect sagaStartAspect(SagaMessageSender sender, OmegaContext context) {
return new SagaStartAspect(sender, context);
}
@Bean
- TransactionAspect transactionAspect(@Qualifier("sagaSender")MessageSender sender, OmegaContext context) {
+ TransactionAspect transactionAspect(SagaMessageSender sender, OmegaContext context) {
return new TransactionAspect(sender, context);
}
@@ -63,21 +63,25 @@ public class TransactionAspectConfig {
@Bean
TccMessageHandler coordinateMessageHandler(
- TccEventService tccEventService,
+ TccMessageSender tccMessageSender,
@Qualifier("coordinateContext") CallbackContext coordinateContext,
OmegaContext omegaContext,
ParametersContext parametersContext) {
- return new CoordinateMessageHandler(tccEventService, coordinateContext, omegaContext, parametersContext);
+ return new CoordinateMessageHandler(tccMessageSender, coordinateContext, omegaContext, parametersContext);
}
@Bean
- TccStartAspect tccStartAspect(TccEventService tccEventService, OmegaContext context) {
- return new TccStartAspect(tccEventService, context);
+ TccStartAspect tccStartAspect(
+ TccMessageSender tccMessageSender,
+ OmegaContext context) {
+ return new TccStartAspect(tccMessageSender, context);
}
@Bean
- TccParticipatorAspect tccParticipatorAspect(TccEventService tccEventService, OmegaContext context, ParametersContext parametersContext) {
- return new TccParticipatorAspect(tccEventService, context, parametersContext);
+ TccParticipatorAspect tccParticipatorAspect(
+ TccMessageSender tccMessageSender,
+ OmegaContext context, ParametersContext parametersContext) {
+ return new TccParticipatorAspect(tccMessageSender, context, parametersContext);
}
@Bean
diff --git a/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/MessageConfig.java b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/MessageConfig.java
index 6230c59..d99bc1b 100644
--- a/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/MessageConfig.java
+++ b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/saga/omega/transaction/spring/MessageConfig.java
@@ -24,10 +24,12 @@ import org.apache.servicecomb.saga.omega.context.IdGenerator;
import org.apache.servicecomb.saga.omega.context.OmegaContext;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
+import org.apache.servicecomb.saga.omega.transaction.SagaMessageSender;
import org.apache.servicecomb.saga.omega.transaction.TxEvent;
import org.apache.servicecomb.saga.omega.transaction.tcc.DefaultParametersContext;
import org.apache.servicecomb.saga.omega.transaction.tcc.ParametersContext;
import org.apache.servicecomb.saga.omega.transaction.tcc.TccEventService;
+import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageSender;
import org.apache.servicecomb.saga.omega.transaction.tcc.events.CoordinatedEvent;
import org.apache.servicecomb.saga.omega.transaction.tcc.events.ParticipatedEvent;
import org.apache.servicecomb.saga.omega.transaction.tcc.events.TccEndedEvent;
@@ -71,8 +73,8 @@ public class MessageConfig {
}
@Bean(name = "sagaSender")
- MessageSender sender() {
- return new MessageSender() {
+ SagaMessageSender sender() {
+ return new SagaMessageSender() {
@Override
public void onConnected() {
@@ -102,8 +104,8 @@ public class MessageConfig {
}
@Bean
- TccEventService tccEventService() {
- return new TccEventService() {
+ TccMessageSender TccMessageSender() {
+ return new TccMessageSender() {
@Override
public void onConnected() {
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/CompensableInterceptor.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/CompensableInterceptor.java
index 588d660..5c5bb18 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/CompensableInterceptor.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/CompensableInterceptor.java
@@ -21,9 +21,9 @@ import org.apache.servicecomb.saga.omega.context.OmegaContext;
class CompensableInterceptor implements EventAwareInterceptor {
private final OmegaContext context;
- private final MessageSender sender;
+ private final SagaMessageSender sender;
- CompensableInterceptor(OmegaContext context, MessageSender sender) {
+ CompensableInterceptor(OmegaContext context, SagaMessageSender sender) {
this.sender = sender;
this.context = context;
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/CompensationMessageHandler.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/CompensationMessageHandler.java
index 75501b9..2e1e0a3 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/CompensationMessageHandler.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/CompensationMessageHandler.java
@@ -20,11 +20,11 @@ package org.apache.servicecomb.saga.omega.transaction;
import org.apache.servicecomb.saga.omega.context.CallbackContext;
public class CompensationMessageHandler implements MessageHandler {
- private final MessageSender sender;
+ private final SagaMessageSender sender;
private final CallbackContext context;
- public CompensationMessageHandler(MessageSender sender, CallbackContext context) {
+ public CompensationMessageHandler(SagaMessageSender sender, CallbackContext context) {
this.sender = sender;
this.context = context;
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/MessageSender.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/MessageSender.java
index 019d8dd..4062082 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/MessageSender.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/MessageSender.java
@@ -6,18 +6,19 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.transaction;
public interface MessageSender {
+
void onConnected();
void onDisconnected();
@@ -26,5 +27,4 @@ public interface MessageSender {
String target();
- AlphaResponse send(TxEvent event);
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/MessageSender.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/SagaMessageSender.java
similarity index 51%
copy from omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/MessageSender.java
copy to omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/SagaMessageSender.java
index 019d8dd..3a6c67c 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/MessageSender.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/SagaMessageSender.java
@@ -6,25 +6,18 @@
* (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
+ * 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.
+ * 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.servicecomb.saga.omega.transaction;
-public interface MessageSender {
- void onConnected();
-
- void onDisconnected();
-
- void close();
-
- String target();
+public interface SagaMessageSender extends MessageSender {
AlphaResponse send(TxEvent event);
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAnnotationProcessor.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAnnotationProcessor.java
index b0b2de4..4c3b122 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAnnotationProcessor.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAnnotationProcessor.java
@@ -23,9 +23,9 @@ import org.apache.servicecomb.saga.omega.context.OmegaContext;
class SagaStartAnnotationProcessor {
private final OmegaContext omegaContext;
- private final MessageSender sender;
+ private final SagaMessageSender sender;
- SagaStartAnnotationProcessor(OmegaContext omegaContext, MessageSender sender) {
+ SagaStartAnnotationProcessor(OmegaContext omegaContext, SagaMessageSender sender) {
this.omegaContext = omegaContext;
this.sender = sender;
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAspect.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAspect.java
index 0dc3279..0e09f49 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAspect.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAspect.java
@@ -36,7 +36,7 @@ public class SagaStartAspect {
private final OmegaContext context;
- public SagaStartAspect(MessageSender sender, OmegaContext context) {
+ public SagaStartAspect(SagaMessageSender sender, OmegaContext context) {
this.context = context;
this.sagaStartAnnotationProcessor = new SagaStartAnnotationProcessor(context, sender);
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/TransactionAspect.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/TransactionAspect.java
index f7a98ee..ee247ce 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/TransactionAspect.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/TransactionAspect.java
@@ -37,7 +37,7 @@ public class TransactionAspect {
private final CompensableInterceptor interceptor;
- public TransactionAspect(MessageSender sender, OmegaContext context) {
+ public TransactionAspect(SagaMessageSender sender, OmegaContext context) {
this.context = context;
this.interceptor = new CompensableInterceptor(context, sender);
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/CoordinateMessageHandler.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/CoordinateMessageHandler.java
index 17b6df8..657c9fb 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/CoordinateMessageHandler.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/CoordinateMessageHandler.java
@@ -24,7 +24,7 @@ import org.apache.servicecomb.saga.omega.transaction.tcc.events.CoordinatedEvent
public class CoordinateMessageHandler implements TccMessageHandler {
- private final TccEventService tccEventService;
+ private final TccMessageSender tccMessageSender;
private final CallbackContext callbackContext;
@@ -32,10 +32,10 @@ public class CoordinateMessageHandler implements TccMessageHandler {
private final ParametersContext parametersContext;
- public CoordinateMessageHandler(TccEventService tccEventService,
+ public CoordinateMessageHandler(TccMessageSender tccMessageSender,
CallbackContext callbackContext, OmegaContext omegaContext,
ParametersContext parametersContext) {
- this.tccEventService = tccEventService;
+ this.tccMessageSender = tccMessageSender;
this.callbackContext = callbackContext;
this.omegaContext = omegaContext;
this.parametersContext = parametersContext;
@@ -46,7 +46,7 @@ public class CoordinateMessageHandler implements TccMessageHandler {
// TODO need to catch the exception and send the failed message
// The parameter need to be updated here
callbackContext.apply(globalTxId, localTxId, methodName, parametersContext.getParameters(localTxId));
- tccEventService.coordinate(new CoordinatedEvent(globalTxId, localTxId, parentTxId, methodName, TransactionStatus.Succeed));
+ tccMessageSender.coordinate(new CoordinatedEvent(globalTxId, localTxId, parentTxId, methodName, TransactionStatus.Succeed));
// Need to remove the parameter
parametersContext.removeParameter(localTxId);
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccMessageSender.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccMessageSender.java
similarity index 96%
rename from omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccMessageSender.java
rename to omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccMessageSender.java
index 00cad14..2a9055e 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/TccMessageSender.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccMessageSender.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
+package org.apache.servicecomb.saga.omega.transaction.tcc;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccParticipatorAspect.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccParticipatorAspect.java
index e3a8fcc..8849660 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccParticipatorAspect.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccParticipatorAspect.java
@@ -35,15 +35,15 @@ public class TccParticipatorAspect {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private final OmegaContext context;
- private final TccEventService tccEventService;
+ private final TccMessageSender tccMessageSender;
private final ParametersContext parametersContext;
//We need to inject the CoordinateMessageHandler for the parameter map
- public TccParticipatorAspect(TccEventService tccEventService, OmegaContext context,
+ public TccParticipatorAspect(TccMessageSender tccMessageSender, OmegaContext context,
ParametersContext parametersContext) {
this.context = context;
- this.tccEventService = tccEventService;
+ this.tccMessageSender = tccMessageSender;
this.parametersContext = parametersContext;
}
@@ -60,7 +60,7 @@ public class TccParticipatorAspect {
try {
Object result = joinPoint.proceed();
// Send the participate message back
- tccEventService.participate(new ParticipatedEvent(context.globalTxId(), context.localTxId(), localTxId, confirmMethod,
+ tccMessageSender.participate(new ParticipatedEvent(context.globalTxId(), context.localTxId(), localTxId, confirmMethod,
cancelMethod, TransactionStatus.Succeed));
// Just store the parameters into the context
parametersContext.putParamters(context.localTxId(), joinPoint.getArgs());
@@ -68,7 +68,7 @@ public class TccParticipatorAspect {
return result;
} catch (Throwable throwable) {
// Now we don't handle the error message
- tccEventService.participate(new ParticipatedEvent(context.globalTxId(), context.localTxId(), localTxId, confirmMethod,
+ tccMessageSender.participate(new ParticipatedEvent(context.globalTxId(), context.localTxId(), localTxId, confirmMethod,
cancelMethod, TransactionStatus.Failed));
LOG.error("Participate Transaction with context {} failed.", context, throwable);
throw throwable;
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAnnotationProcessor.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAnnotationProcessor.java
index 137a397..1d4cf24 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAnnotationProcessor.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAnnotationProcessor.java
@@ -17,7 +17,6 @@
package org.apache.servicecomb.saga.omega.transaction.tcc;
import javax.transaction.TransactionalException;
-
import org.apache.servicecomb.saga.common.TransactionStatus;
import org.apache.servicecomb.saga.omega.context.OmegaContext;
import org.apache.servicecomb.saga.omega.transaction.AlphaResponse;
@@ -28,30 +27,30 @@ import org.apache.servicecomb.saga.omega.transaction.tcc.events.TccStartedEvent;
public class TccStartAnnotationProcessor {
private final OmegaContext omegaContext;
- private final TccEventService eventService;
+ private final TccMessageSender tccMessageSender;
- TccStartAnnotationProcessor(OmegaContext omegaContext, TccEventService eventService) {
+ TccStartAnnotationProcessor(OmegaContext omegaContext, TccMessageSender tccMessageSender) {
this.omegaContext = omegaContext;
- this.eventService = eventService;
+ this.tccMessageSender = tccMessageSender;
}
public AlphaResponse preIntercept(String parentTxId, String methodName, int timeout) {
try {
- return eventService.tccTransactionStart(new TccStartedEvent(omegaContext.globalTxId(), omegaContext.localTxId()));
+ return tccMessageSender.tccTransactionStart(new TccStartedEvent(omegaContext.globalTxId(), omegaContext.localTxId()));
} catch (OmegaException e) {
throw new TransactionalException(e.getMessage(), e.getCause());
}
}
public void postIntercept(String parentTxId, String methodName) {
- eventService.tccTransactionStop(new TccEndedEvent(omegaContext.globalTxId(), omegaContext.localTxId(),
+ tccMessageSender.tccTransactionStop(new TccEndedEvent(omegaContext.globalTxId(), omegaContext.localTxId(),
TransactionStatus.Succeed));
}
public void onError(String parentTxId, String methodName, Throwable throwable) {
// Send the cancel event
// Do we need to wait for the alpha finish all the transaction
- eventService.tccTransactionStop(new TccEndedEvent(omegaContext.globalTxId(), omegaContext.localTxId(),
+ tccMessageSender.tccTransactionStop(new TccEndedEvent(omegaContext.globalTxId(), omegaContext.localTxId(),
TransactionStatus.Failed));
}
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAspect.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAspect.java
index 62a2873..8d0bf83 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAspect.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAspect.java
@@ -37,9 +37,9 @@ public class TccStartAspect {
private final OmegaContext context;
- public TccStartAspect(TccEventService tccEventServicer, OmegaContext context) {
+ public TccStartAspect(TccMessageSender tccMessageSender, OmegaContext context) {
this.context = context;
- this.tccStartAnnotationProcessor = new TccStartAnnotationProcessor(context, tccEventServicer);
+ this.tccStartAnnotationProcessor = new TccStartAnnotationProcessor(context, tccMessageSender);
}
@Around("execution(@org.apache.servicecomb.saga.omega.context.annotations.TccStart * *(..)) && @annotation(tccStart)")
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/CompensableInterceptorTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/CompensableInterceptorTest.java
index aefc9a4..aec9d5a 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/CompensableInterceptorTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/CompensableInterceptorTest.java
@@ -42,7 +42,7 @@ public class CompensableInterceptorTest {
private final String localTxId = UUID.randomUUID().toString();
private final String parentTxId = UUID.randomUUID().toString();
- private final MessageSender sender =new MessageSender() {
+ private final SagaMessageSender sender = new SagaMessageSender() {
@Override
public void onConnected() {
}
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/CompensationMessageHandlerTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/CompensationMessageHandlerTest.java
index b9edb07..ce40d71 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/CompensationMessageHandlerTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/CompensationMessageHandlerTest.java
@@ -34,7 +34,7 @@ import org.junit.Test;
public class CompensationMessageHandlerTest {
private final List<TxEvent> events = new ArrayList<>();
- private final MessageSender sender = new MessageSender() {
+ private final SagaMessageSender sender = new SagaMessageSender() {
@Override
public void onConnected() {
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/DefaultRecoveryTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/DefaultRecoveryTest.java
index edbabb9..3f8b17f 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/DefaultRecoveryTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/DefaultRecoveryTest.java
@@ -67,7 +67,7 @@ public class DefaultRecoveryTest {
private final Compensable compensable = mock(Compensable.class);
- private final MessageSender sender = new MessageSender() {
+ private final SagaMessageSender sender = new SagaMessageSender() {
@Override
public void onConnected() {
@@ -165,7 +165,7 @@ public class DefaultRecoveryTest {
@Test
public void returnImmediatelyWhenReceivedRejectResponse() {
- MessageSender sender = mock(MessageSender.class);
+ SagaMessageSender sender = mock(SagaMessageSender.class);
when(sender.send(any(TxEvent.class))).thenReturn(new AlphaResponse(true));
CompensableInterceptor interceptor = new CompensableInterceptor(omegaContext, sender);
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/ForwardRecoveryTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/ForwardRecoveryTest.java
index 9ba4b47..98b3980 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/ForwardRecoveryTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/ForwardRecoveryTest.java
@@ -66,7 +66,7 @@ public class ForwardRecoveryTest {
private final Compensable compensable = mock(Compensable.class);
- private final MessageSender sender = new MessageSender() {
+ private final SagaMessageSender sender = new SagaMessageSender() {
@Override
public void onConnected() {
@@ -116,7 +116,7 @@ public class ForwardRecoveryTest {
@Test
public void forwardExceptionWhenGlobalTxAborted() {
- MessageSender sender = mock(MessageSender.class);
+ SagaMessageSender sender = mock(SagaMessageSender.class);
when(sender.send(any(TxEvent.class))).thenReturn(new AlphaResponse(true));
CompensableInterceptor interceptor = new CompensableInterceptor(omegaContext, sender);
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAnnotationProcessorTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAnnotationProcessorTest.java
index 6d50e6f..9052e2f 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAnnotationProcessorTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAnnotationProcessorTest.java
@@ -40,7 +40,7 @@ public class SagaStartAnnotationProcessorTest {
private final List<TxEvent> messages = new ArrayList<>();
- private final MessageSender sender = new MessageSender() {
+ private final SagaMessageSender sender = new SagaMessageSender() {
@Override
public void onConnected() {
@@ -116,7 +116,7 @@ public class SagaStartAnnotationProcessorTest {
@Test
public void transformInterceptedException() {
- MessageSender sender = mock(MessageSender.class);
+ SagaMessageSender sender = mock(SagaMessageSender.class);
SagaStartAnnotationProcessor sagaStartAnnotationProcessor = new SagaStartAnnotationProcessor(
context, sender);
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAspectTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAspectTest.java
index 4025c1c..a9ff338 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAspectTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/SagaStartAspectTest.java
@@ -35,14 +35,13 @@ import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.reflect.MethodSignature;
import org.junit.Before;
import org.junit.Test;
-import org.junit.rules.ExpectedException;
import org.mockito.Mockito;
public class SagaStartAspectTest {
private final List<TxEvent> messages = new ArrayList<>();
private final String globalTxId = UUID.randomUUID().toString();
- private final MessageSender sender = new MessageSender() {
+ private final SagaMessageSender sender = new SagaMessageSender() {
@Override
public void onConnected() {
}
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/TransactionAspectTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/TransactionAspectTest.java
index d7fade6..3c05140 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/TransactionAspectTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/TransactionAspectTest.java
@@ -42,7 +42,7 @@ public class TransactionAspectTest {
private final String localTxId = UUID.randomUUID().toString();
private final String newLocalTxId = UUID.randomUUID().toString();
- private final MessageSender sender = new MessageSender() {
+ private final SagaMessageSender sender = new SagaMessageSender() {
@Override
public void onConnected() {
}
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/CoordinateMessageHandlerTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/CoordinateMessageHandlerTest.java
index e2e0769..6dbb721 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/CoordinateMessageHandlerTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/CoordinateMessageHandlerTest.java
@@ -24,7 +24,6 @@ import static org.junit.Assert.assertThat;
import java.util.ArrayList;
import java.util.List;
-
import org.apache.servicecomb.saga.common.TransactionStatus;
import org.apache.servicecomb.saga.omega.context.CallbackContext;
import org.apache.servicecomb.saga.omega.context.OmegaContext;
@@ -35,13 +34,12 @@ import org.apache.servicecomb.saga.omega.transaction.tcc.events.TccEndedEvent;
import org.apache.servicecomb.saga.omega.transaction.tcc.events.TccStartedEvent;
import org.junit.Before;
import org.junit.Test;
-import org.mockito.Mock;
import org.mockito.Mockito;
public class CoordinateMessageHandlerTest {
private final List<CoordinatedEvent> coordinatedEvents = new ArrayList<>();
private final AlphaResponse response = new AlphaResponse(false);
- private final TccEventService eventService = new TccEventService() {
+ private final TccMessageSender tccMessageSender = new TccMessageSender() {
@Override
public void onConnected() {
@@ -92,7 +90,7 @@ public class CoordinateMessageHandlerTest {
private final CallbackContext callbackContext = Mockito.mock(CallbackContext.class);
private final OmegaContext omegaContext = Mockito.mock(OmegaContext.class);
private final ParametersContext parametersContext = Mockito.mock(ParametersContext.class);
- private final CoordinateMessageHandler handler = new CoordinateMessageHandler(eventService, callbackContext, omegaContext, parametersContext);
+ private final CoordinateMessageHandler handler = new CoordinateMessageHandler(tccMessageSender, callbackContext, omegaContext, parametersContext);
@Before
public void setUp() {
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccParticipatorAspectTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccParticipatorAspectTest.java
index d308876..f52da03 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccParticipatorAspectTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccParticipatorAspectTest.java
@@ -52,7 +52,7 @@ public class TccParticipatorAspectTest {
private final AlphaResponse response = new AlphaResponse(false);
private String confirmMethod;
private String cancelMethod;
- private final TccEventService eventService = new TccEventService() {
+ private final TccMessageSender tccMessageSender = new TccMessageSender() {
@Override
public void onConnected() {
@@ -103,7 +103,7 @@ public class TccParticipatorAspectTest {
private final Participate participate = mock(Participate.class);
private final ParametersContext parametersContext = new DefaultParametersContext();
- private final TccParticipatorAspect aspect = new TccParticipatorAspect(eventService, omegaContext, parametersContext);
+ private final TccParticipatorAspect aspect = new TccParticipatorAspect(tccMessageSender, omegaContext, parametersContext);
@Before
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAnnotationProcessorTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAnnotationProcessorTest.java
index fa34ad7..1853874 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAnnotationProcessorTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAnnotationProcessorTest.java
@@ -53,7 +53,7 @@ public class TccStartAnnotationProcessorTest {
private final IdGenerator<String> generator = mock(IdGenerator.class);
private final OmegaContext context = new OmegaContext(generator);
private final OmegaException exception = new OmegaException("exception", new RuntimeException("runtime exception"));
- private final TccEventService eventService = new TccEventService() {
+ private final TccMessageSender tccMessageSender = new TccMessageSender() {
@Override
public void onConnected() {
@@ -100,7 +100,7 @@ public class TccStartAnnotationProcessorTest {
}
};
private final TccStartAnnotationProcessor tccStartAnnotationProcessor = new TccStartAnnotationProcessor(context,
- eventService);
+ tccMessageSender);
@Before
public void setUp() throws Exception {
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAspectTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAspectTest.java
index 80f14e4..0fe096d 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAspectTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAspectTest.java
@@ -49,7 +49,7 @@ public class TccStartAspectTest {
private final List<TccStartedEvent> startedEvents = new ArrayList<>();
private final List<TccEndedEvent> endedEvents = new ArrayList<>();
private final AlphaResponse response = new AlphaResponse(false);
- private final TccEventService eventService = new TccEventService() {
+ private final TccMessageSender tccMessageSender = new TccMessageSender() {
@Override
public void onConnected() {
@@ -99,7 +99,7 @@ public class TccStartAspectTest {
private final TccStart tccStart = Mockito.mock(TccStart.class);
private final OmegaContext omegaContext = new OmegaContext(idGenerator);
- private final TccStartAspect aspect = new TccStartAspect(eventService, omegaContext);
+ private final TccStartAspect aspect = new TccStartAspect(tccMessageSender, omegaContext);
@Before
public void setUp() throws Exception {