diff options
Diffstat (limited to 'models-interactions')
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); |