X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=policy-agent%2Fsrc%2Fmain%2Fjava%2Forg%2Foransc%2Fpolicyagent%2Fdmaap%2FDmaapMessageHandler.java;h=c50a46fa091ec39f6733b3930c2ab74ff45e92f4;hb=ad3dbd1a9bb1f0a9d5f90be7e5dfe835ae6c1a41;hp=fc6e439a3884e8d31778f82fee98bdb6f4883b14;hpb=e2ac0dcf30eaf828a7c7bbd722730410f4110030;p=nonrtric.git diff --git a/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageHandler.java b/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageHandler.java index fc6e439a..c50a46fa 100644 --- a/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageHandler.java +++ b/policy-agent/src/main/java/org/oransc/policyagent/dmaap/DmaapMessageHandler.java @@ -17,6 +17,7 @@ * limitations under the License. * ========================LICENSE_END=================================== */ + package org.oransc.policyagent.dmaap; import com.google.gson.Gson; @@ -56,10 +57,12 @@ public class DmaapMessageHandler { } public void handleDmaapMsg(String msg) { - this.createTask(msg) // - .subscribe(message -> logger.debug("handleDmaapMsg: {}", message), // - throwable -> logger.warn("handleDmaapMsg failure ", throwable), // - () -> logger.debug("handleDmaapMsg complete")); + try { + String result = this.createTask(msg).block(); + logger.debug("handleDmaapMsg: {}", result); + } catch (Exception throwable) { + logger.warn("handleDmaapMsg failure {}", throwable.getMessage()); + } } Mono createTask(String msg) { @@ -70,8 +73,8 @@ public class DmaapMessageHandler { .flatMap( response -> sendDmaapResponse(response.getBody(), dmaapRequestMessage, response.getStatusCode())); } catch (Exception e) { - logger.warn("Received unparsable message from DMAAP: {}", msg); - return Mono.error(e); // Cannot make any response + String errorMsg = "Received unparsable message from DMAAP: \"" + msg + "\", reason: " + e.getMessage(); + return Mono.error(new ServiceException(errorMsg)); // Cannot make any response } }