summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperation.java15
-rw-r--r--models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperationTest.java32
2 files changed, 41 insertions, 6 deletions
diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperation.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperation.java
index ec522a405..f598d627a 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperation.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperation.java
@@ -106,7 +106,7 @@ public abstract class BidirectionalTopicOperation<Q, S> extends OperationPartial
@Override
protected CompletableFuture<OperationOutcome> startOperationAsync(int attempt, OperationOutcome outcome) {
- final Pair<String,Q> pair = makeRequest(attempt);
+ final Pair<String, Q> pair = makeRequest(attempt);
final Q request = pair.getRight();
outcome.setSubRequestId(pair.getLeft());
@@ -118,10 +118,15 @@ public abstract class BidirectionalTopicOperation<Q, S> extends OperationPartial
// register a listener BEFORE publishing
BiConsumer<String, StandardCoderObject> listener = (rawResponse, scoResponse) -> {
- OperationOutcome latestOutcome = processResponse(outcome, rawResponse, scoResponse);
- if (latestOutcome != null) {
- // final response - complete the controller
- controller.completeAsync(() -> latestOutcome, executor);
+ try {
+ OperationOutcome latestOutcome = processResponse(outcome, rawResponse, scoResponse);
+ if (latestOutcome != null) {
+ // final response - complete the controller
+ controller.completeAsync(() -> latestOutcome, executor);
+ }
+ } catch (RuntimeException e) {
+ logger.warn("{}: failed to process response for {}", getFullName(), params.getRequestId());
+ controller.completeExceptionally(e);
}
};
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperationTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperationTest.java
index 587564a2e..48669f799 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperationTest.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperationTest.java
@@ -168,10 +168,40 @@ public class BidirectionalTopicOperationTest {
}
/**
+ * Tests startOperationAsync() when processResponse() throws an exception.
+ */
+ @Test
+ public void testStartOperationAsyncProcException() throws Exception {
+ oper = new MyOperation() {
+ @Override
+ protected OperationOutcome processResponse(OperationOutcome outcome, String rawResponse,
+ StandardCoderObject scoResponse) {
+ throw EXPECTED_EXCEPTION;
+ }
+ };
+
+ CompletableFuture<OperationOutcome> future = oper.startOperationAsync(1, outcome);
+ assertFalse(future.isDone());
+
+ assertEquals(SUB_REQID, outcome.getSubRequestId());
+
+ verify(forwarder).register(eq(Arrays.asList(REQ_ID)), listenerCaptor.capture());
+
+ verify(forwarder, never()).unregister(any(), any());
+
+ // provide a response
+ listenerCaptor.getValue().accept(responseText, stdResponse);
+ assertTrue(executor.runAll(MAX_REQUESTS));
+ assertTrue(future.isCompletedExceptionally());
+
+ verify(forwarder).unregister(eq(Arrays.asList(REQ_ID)), eq(listenerCaptor.getValue()));
+ }
+
+ /**
* Tests startOperationAsync() when the publisher throws an exception.
*/
@Test
- public void testStartOperationAsyncException() throws Exception {
+ public void testStartOperationAsyncPubException() throws Exception {
// indicate that nothing was published
when(handler.send(any())).thenReturn(false);