You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicecomb.apache.org by se...@apache.org on 2018/01/04 08:03:43 UTC
[incubator-servicecomb-saga] 01/07: SCB-138 set up bidirectional
streaming for alpha
This is an automated email from the ASF dual-hosted git repository.
seanyinx pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-saga.git
commit 31e91ba817dca1f7d5569715e46db8fe97b14ccc
Author: Eric Lee <da...@huawei.com>
AuthorDate: Tue Jan 2 17:49:21 2018 +0800
SCB-138 set up bidirectional streaming for alpha
Signed-off-by: Eric Lee <da...@huawei.com>
---
.../servicecomb/saga/alpha/server/AlphaConfig.java | 28 +++----
.../saga/alpha/server/GrpcOmegaCallback.java | 49 ++++++++++++
.../saga/alpha/server/GrpcOmegaStreamObserver.java | 90 ++++++++++++++++++++++
.../saga/alpha/server/GrpcTxEventEndpointImpl.java | 33 +++-----
.../saga/alpha/server/AlphaIntegrationTest.java | 49 +++++++++++-
.../src/main/proto/GrpcTxEvent.proto | 10 ++-
6 files changed, 222 insertions(+), 37 deletions(-)
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/AlphaConfig.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/AlphaConfig.java
index 7bd855e..e33ba99 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/AlphaConfig.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/AlphaConfig.java
@@ -54,25 +54,27 @@ class AlphaConfig {
}
@Bean
- TxEventRepository springTxEventRepository(@Value("${alpha.server.port:8080}") int port,
- TxEventEnvelopeRepository eventRepo,
- OmegaCallback omegaCallback) {
+ TxEventRepository springTxEventRepository(TxEventEnvelopeRepository eventRepo) {
+ return new SpringTxEventRepository(eventRepo);
+ }
+
+ @Bean
+ TxConsistentService txConsistentService(@Value("${alpha.server.port:8080}") int port,
+ TxEventRepository eventRepository,
+ OmegaCallback omegaCallback,
+ Map<String, Map<String, OmegaCallback>> omegaCallbacks) {
- TxEventRepository eventRepository = new SpringTxEventRepository(eventRepo);
+ TxConsistentService consistentService = new TxConsistentService(eventRepository, omegaCallback);
- ServerStartable startable = buildGrpc(port, omegaCallback, eventRepository);
+ ServerStartable startable = buildGrpc(port, consistentService, omegaCallbacks);
new Thread(startable::start).start();
- return eventRepository;
+ return consistentService;
}
- private ServerStartable buildGrpc(int port, OmegaCallback omegaCallback, TxEventRepository eventRepository) {
- return new GrpcStartable(
- port,
- new GrpcTxEventEndpointImpl(
- new TxConsistentService(
- eventRepository,
- omegaCallback)));
+ private ServerStartable buildGrpc(int port, TxConsistentService txConsistentService,
+ Map<String, Map<String, OmegaCallback>> omegaCallbacks) {
+ return new GrpcStartable(port, new GrpcTxEventEndpointImpl(txConsistentService, omegaCallbacks));
}
@PostConstruct
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/GrpcOmegaCallback.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/GrpcOmegaCallback.java
new file mode 100644
index 0000000..db04253
--- /dev/null
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/GrpcOmegaCallback.java
@@ -0,0 +1,49 @@
+/*
+ *
+ * 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.alpha.server;
+
+import org.apache.servicecomb.saga.alpha.core.OmegaCallback;
+import org.apache.servicecomb.saga.alpha.core.TxEvent;
+import org.apache.servicecomb.saga.pack.contract.grpc.GrpcCompensateCommand;
+
+import com.google.protobuf.ByteString;
+
+import io.grpc.stub.StreamObserver;
+
+public class GrpcOmegaCallback implements OmegaCallback {
+
+ private final StreamObserver<GrpcCompensateCommand> observer;
+
+ public GrpcOmegaCallback(StreamObserver<GrpcCompensateCommand> observer) {
+ this.observer = observer;
+ }
+
+ @Override
+ public void compensate(TxEvent event) {
+ GrpcCompensateCommand command = GrpcCompensateCommand.newBuilder()
+ .setGlobalTxId(event.globalTxId())
+ .setLocalTxId(event.localTxId())
+ .setCompensateMethod(event.compensationMethod())
+ .setPayloads(ByteString.copyFrom(event.payloads()))
+ .build();
+ observer.onNext(command);
+ }
+}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/GrpcOmegaStreamObserver.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/GrpcOmegaStreamObserver.java
new file mode 100644
index 0000000..d8d10f5
--- /dev/null
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/GrpcOmegaStreamObserver.java
@@ -0,0 +1,90 @@
+/*
+ *
+ * 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.alpha.server;
+
+import static org.apache.servicecomb.saga.alpha.core.EventType.TxStartedEvent;
+
+import java.lang.invoke.MethodHandles;
+import java.util.Date;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.servicecomb.saga.alpha.core.OmegaCallback;
+import org.apache.servicecomb.saga.alpha.core.TxConsistentService;
+import org.apache.servicecomb.saga.alpha.core.TxEvent;
+import org.apache.servicecomb.saga.pack.contract.grpc.GrpcCompensateCommand;
+import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTxEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.grpc.stub.StreamObserver;
+
+class GrpcOmegaStreamObserver implements StreamObserver<GrpcTxEvent> {
+
+ private static Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private final Map<String, Map<String, OmegaCallback>> omegaCallbacks;
+
+ private final TxConsistentService txConsistentService;
+
+ private final StreamObserver<GrpcCompensateCommand> responseObserver;
+
+ GrpcOmegaStreamObserver(Map<String, Map<String, OmegaCallback>> omegaCallbacks,
+ TxConsistentService txConsistentService, StreamObserver<GrpcCompensateCommand> responseObserver) {
+ this.omegaCallbacks = omegaCallbacks;
+ this.txConsistentService = txConsistentService;
+ this.responseObserver = responseObserver;
+ }
+
+ @Override
+ public void onNext(GrpcTxEvent message) {
+ // register a callback on started event
+ if (message.getType().equals(TxStartedEvent.name())) {
+ Map<String, OmegaCallback> callbacks = new ConcurrentHashMap<>();
+ callbacks.put(message.getInstanceId(), new GrpcOmegaCallback(responseObserver));
+ omegaCallbacks.put(message.getServiceName(), callbacks);
+ }
+
+ // store received event
+ txConsistentService.handle(new TxEvent(
+ message.getServiceName(),
+ message.getInstanceId(),
+ new Date(message.getTimestamp()),
+ message.getGlobalTxId(),
+ message.getLocalTxId(),
+ message.getParentTxId().isEmpty() ? null : message.getParentTxId(),
+ message.getType(),
+ message.getCompensationMethod(),
+ message.getPayloads().toByteArray()
+ ));
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ LOG.error("failed to process grpc message.", t);
+ responseObserver.onCompleted();
+ }
+
+ @Override
+ public void onCompleted() {
+ responseObserver.onCompleted();
+ }
+}
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/GrpcTxEventEndpointImpl.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/GrpcTxEventEndpointImpl.java
index cba9f0c..b68972e 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/GrpcTxEventEndpointImpl.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/GrpcTxEventEndpointImpl.java
@@ -20,39 +20,30 @@
package org.apache.servicecomb.saga.alpha.server;
-import java.util.Date;
+import java.util.Map;
-import io.grpc.stub.StreamObserver;
+import org.apache.servicecomb.saga.alpha.core.OmegaCallback;
import org.apache.servicecomb.saga.alpha.core.TxConsistentService;
-import org.apache.servicecomb.saga.alpha.core.TxEvent;
-import org.apache.servicecomb.saga.pack.contract.grpc.GrpcEmpty;
+import org.apache.servicecomb.saga.pack.contract.grpc.GrpcCompensateCommand;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTxEvent;
import org.apache.servicecomb.saga.pack.contract.grpc.TxEventServiceGrpc.TxEventServiceImplBase;
+import io.grpc.stub.StreamObserver;
+
class GrpcTxEventEndpointImpl extends TxEventServiceImplBase {
private final TxConsistentService txConsistentService;
- GrpcTxEventEndpointImpl(TxConsistentService txConsistentService) {
+ private final Map<String, Map<String, OmegaCallback>> omegaCallbacks;
+
+ GrpcTxEventEndpointImpl(TxConsistentService txConsistentService,
+ Map<String, Map<String, OmegaCallback>> omegaCallbacks) {
this.txConsistentService = txConsistentService;
+ this.omegaCallbacks = omegaCallbacks;
}
@Override
- public void reportEvent(GrpcTxEvent message, StreamObserver<GrpcEmpty> responseObserver) {
- txConsistentService.handle(new TxEvent(
- message.getServiceName(),
- message.getInstanceId(),
- new Date(message.getTimestamp()),
- message.getGlobalTxId(),
- message.getLocalTxId(),
- message.getParentTxId().isEmpty()? null : message.getParentTxId(),
- message.getType(),
- message.getCompensationMethod(),
- message.getPayloads().toByteArray()
- ));
-
- GrpcEmpty reply = GrpcEmpty.newBuilder().build();
- responseObserver.onNext(reply);
- responseObserver.onCompleted();
+ public StreamObserver<GrpcTxEvent> callbackCommand(StreamObserver<GrpcCompensateCommand> responseObserver) {
+ return new GrpcOmegaStreamObserver(omegaCallbacks, txConsistentService, responseObserver);
}
}
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/AlphaIntegrationTest.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/AlphaIntegrationTest.java
index 221c5b0..20921a8 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/AlphaIntegrationTest.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/AlphaIntegrationTest.java
@@ -28,6 +28,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
import java.util.UUID;
@@ -48,11 +49,24 @@ import com.google.protobuf.ByteString;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
+<<<<<<< HEAD:alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/AlphaIntegrationTest.java
import org.apache.servicecomb.saga.alpha.server.AlphaIntegrationTest.OmegaCallbackConfig;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTxEvent;
import org.apache.servicecomb.saga.pack.contract.grpc.TxEventServiceGrpc;
import org.apache.servicecomb.saga.pack.contract.grpc.TxEventServiceGrpc.TxEventServiceBlockingStub;
+=======
+import io.grpc.stub.StreamObserver;
+import io.servicecomb.saga.alpha.core.EventType;
+import io.servicecomb.saga.alpha.core.OmegaCallback;
+import io.servicecomb.saga.alpha.core.TxConsistentService;
+import io.servicecomb.saga.alpha.core.TxEvent;
+import io.servicecomb.saga.alpha.server.AlphaIntegrationTest.OmegaCallbackConfig;
+import io.servicecomb.saga.pack.contract.grpc.GrpcCompensateCommand;
+import io.servicecomb.saga.pack.contract.grpc.GrpcTxEvent;
+import io.servicecomb.saga.pack.contract.grpc.TxEventServiceGrpc;
+import io.servicecomb.saga.pack.contract.grpc.TxEventServiceGrpc.TxEventServiceStub;
+>>>>>>> 6985eb8... SCB-138 set up bidirectional streaming for alpha:alpha/alpha-server/src/test/java/io/servicecomb/saga/alpha/server/AlphaIntegrationTest.java
@RunWith(SpringRunner.class)
@SpringBootTest(classes = {AlphaApplication.class, OmegaCallbackConfig.class}, properties = "alpha.server.port=8090")
@@ -62,7 +76,7 @@ public class AlphaIntegrationTest {
private static ManagedChannel clientChannel = ManagedChannelBuilder
.forAddress("localhost", port).usePlaintext(true).build();
- private TxEventServiceBlockingStub stub = TxEventServiceGrpc.newBlockingStub(clientChannel);
+ private TxEventServiceStub stub = TxEventServiceGrpc.newStub(clientChannel);
private static final String payload = "hello world";
@@ -79,6 +93,27 @@ public class AlphaIntegrationTest {
@Autowired
private List<CompensationContext> compensationContexts;
+ @Autowired
+ private Map<String, Map<String, OmegaCallback>> omegaCallbacks;
+
+ @Autowired
+ private TxConsistentService txConsistentService;
+
+ // use an empty response observer as we don't need the response
+ private final StreamObserver<GrpcCompensateCommand> emptyResponseObserver = new StreamObserver<GrpcCompensateCommand>() {
+ @Override
+ public void onNext(GrpcCompensateCommand value) {
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ }
+
+ @Override
+ public void onCompleted() {
+ }
+ };
+
@AfterClass
public static void tearDown() throws Exception {
clientChannel.shutdown();
@@ -86,7 +121,14 @@ public class AlphaIntegrationTest {
@Test
public void persistsEvent() throws Exception {
+<<<<<<< HEAD:alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/AlphaIntegrationTest.java
stub.reportEvent(someGrpcEvent(EventType.TxStartedEvent));
+=======
+ StreamObserver<GrpcTxEvent> requestObserver = stub.callbackCommand(emptyResponseObserver);
+ requestObserver.onNext(someGrpcEvent(TxStartedEvent));
+ // use the asynchronous stub need to wait for some time
+ await().atMost(1, SECONDS).until(() -> eventRepo.findByEventGlobalTxId(globalTxId) != null);
+>>>>>>> 6985eb8... SCB-138 set up bidirectional streaming for alpha:alpha/alpha-server/src/test/java/io/servicecomb/saga/alpha/server/AlphaIntegrationTest.java
TxEventEnvelope envelope = eventRepo.findByEventGlobalTxId(globalTxId);
@@ -111,7 +153,12 @@ public class AlphaIntegrationTest {
eventRepo.save(eventEnvelopeOf(EventType.TxStartedEvent, localTxId1, UUID.randomUUID().toString(), "service b".getBytes(), "method b"));
eventRepo.save(eventEnvelopeOf(EventType.TxEndedEvent, new byte[0], "method b"));
+<<<<<<< HEAD:alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/server/AlphaIntegrationTest.java
stub.reportEvent(someGrpcEvent(EventType.TxAbortedEvent));
+=======
+ StreamObserver<GrpcTxEvent> requestObserver = stub.callbackCommand(emptyResponseObserver);
+ requestObserver.onNext(someGrpcEvent(TxAbortedEvent));
+>>>>>>> 6985eb8... SCB-138 set up bidirectional streaming for alpha:alpha/alpha-server/src/test/java/io/servicecomb/saga/alpha/server/AlphaIntegrationTest.java
await().atMost(1, SECONDS).until(() -> compensationContexts.size() > 1);
assertThat(compensationContexts, containsInAnyOrder(
diff --git a/pack-contracts/pack-contract-grpc/src/main/proto/GrpcTxEvent.proto b/pack-contracts/pack-contract-grpc/src/main/proto/GrpcTxEvent.proto
index 0679637..75a5664 100644
--- a/pack-contracts/pack-contract-grpc/src/main/proto/GrpcTxEvent.proto
+++ b/pack-contracts/pack-contract-grpc/src/main/proto/GrpcTxEvent.proto
@@ -22,7 +22,8 @@ option java_package = "org.apache.servicecomb.saga.pack.contract.grpc";
option java_outer_classname = "TxEventProto";
service TxEventService {
- rpc ReportEvent (GrpcTxEvent) returns (GrpcEmpty) {}
+ rpc CallbackCommand (stream GrpcTxEvent) returns (stream GrpcCompensateCommand) {
+ }
}
message GrpcTxEvent {
@@ -37,4 +38,9 @@ message GrpcTxEvent {
string instanceId = 9;
}
-message GrpcEmpty {}
+message GrpcCompensateCommand {
+ string globalTxId = 1;
+ string localTxId = 2;
+ string compensateMethod = 3;
+ bytes payloads = 4;
+}
\ No newline at end of file
--
To stop receiving notification emails like this one, please contact
"commits@servicecomb.apache.org" <co...@servicecomb.apache.org>.