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=19d156488c679b6e807dc9abf81e054c5e1efc8d;hb=5f72ba4fab7f2a8bf54afd41847774e373253f7d;hp=a158b400bf694bfc250008fb16671b651c41941e;hpb=0b2f51f6215dad3fc93d74572ad23f950ad8ae98;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 a158b400..19d15648 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 @@ -2,7 +2,7 @@ * ========================LICENSE_START================================= * O-RAN-SC * %% - * Copyright (C) 2019 Nordix Foundation + * Copyright (C) 2020 Nordix Foundation * %% * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -22,120 +22,153 @@ package org.oransc.policyagent.dmaap; import com.google.gson.Gson; import com.google.gson.GsonBuilder; +import com.google.gson.JsonObject; import java.io.IOException; +import java.util.Optional; import org.onap.dmaap.mr.client.MRBatchingPublisher; import org.oransc.policyagent.clients.AsyncRestClient; -import org.oransc.policyagent.configuration.ApplicationConfig; import org.oransc.policyagent.dmaap.DmaapRequestMessage.Operation; +import org.oransc.policyagent.exceptions.ServiceException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; +import org.springframework.web.reactive.function.client.WebClientException; +import org.springframework.web.reactive.function.client.WebClientResponseException; import reactor.core.publisher.Mono; +/** + * The class handles incoming requests from DMAAP. + *

+ * That means: invoke a REST call towards this services and to send back a + * response though DMAAP + */ public class DmaapMessageHandler { - private static final Logger logger = LoggerFactory.getLogger(DmaapMessageHandler.class); - private static Gson gson = new GsonBuilder() // - .serializeNulls() // .create(); // - private final MRBatchingPublisher dmaapClient; private final AsyncRestClient agentClient; - public DmaapMessageHandler(MRBatchingPublisher dmaapClient, ApplicationConfig applicationConfig, - AsyncRestClient agentClient) { + public DmaapMessageHandler(MRBatchingPublisher dmaapClient, AsyncRestClient agentClient) { this.agentClient = agentClient; this.dmaapClient = dmaapClient; } public void handleDmaapMsg(String msg) { try { - this.createTask(msg) // - .subscribe(x -> logger.debug("handleDmaapMsg: " + x), // - throwable -> logger.warn("handleDmaapMsg failure ", throwable), // - () -> logger.debug("handleDmaapMsg complete")); - } catch (Exception e) { - logger.warn("Received unparsable message from DMAAP: {}", msg); + String result = this.createTask(msg).block(); + logger.debug("handleDmaapMsg: {}", result); + } catch (Exception throwable) { + logger.warn("handleDmaapMsg failure {}", throwable.getMessage()); } } Mono createTask(String msg) { try { DmaapRequestMessage dmaapRequestMessage = gson.fromJson(msg, ImmutableDmaapRequestMessage.class); - return this.invokePolicyAgent(dmaapRequestMessage) // - .onErrorResume(t -> handleAgentCallError(t, dmaapRequestMessage)) // - .flatMap(response -> sendDmaapResponse(response, dmaapRequestMessage, HttpStatus.OK)); - + .onErrorResume(t -> handleAgentCallError(t, msg, dmaapRequestMessage)) // + .flatMap( + response -> sendDmaapResponse(response.getBody(), dmaapRequestMessage, response.getStatusCode())); } catch (Exception e) { - logger.warn("Received unparsable message from DMAAP: {}", msg); - return Mono.error(e); + String errorMsg = "Received unparsable message from DMAAP: \"" + msg + "\", reason: " + e.getMessage(); + return Mono.error(new ServiceException(errorMsg)); // Cannot make any response + } + } + + private Mono> handleAgentCallError(Throwable t, String originalMessage, + DmaapRequestMessage dmaapRequestMessage) { + logger.debug("Agent call failed: {}", t.getMessage()); + HttpStatus status = HttpStatus.INTERNAL_SERVER_ERROR; + String errorMessage = t.getMessage(); + if (t instanceof WebClientResponseException) { + WebClientResponseException exception = (WebClientResponseException) t; + status = exception.getStatusCode(); + errorMessage = exception.getResponseBodyAsString(); + } else if (t instanceof ServiceException) { + status = HttpStatus.BAD_REQUEST; + errorMessage = prepareBadOperationErrorMessage(t, originalMessage); + } else if (!(t instanceof WebClientException)) { + logger.warn("Unexpected exception ", t); } + return sendDmaapResponse(errorMessage, dmaapRequestMessage, status) // + .flatMap(notUsed -> Mono.empty()); } - private Mono handleAgentCallError(Throwable t, DmaapRequestMessage dmaapRequestMessage) { - logger.debug("Agent call failed: " + t.getMessage()); - return sendDmaapResponse(t.toString(), dmaapRequestMessage, HttpStatus.NOT_FOUND) // - .flatMap(s -> Mono.empty()); + private String prepareBadOperationErrorMessage(Throwable t, String originalMessage) { + String operationParameterStart = "operation\":\""; + int indexOfOperationStart = originalMessage.indexOf(operationParameterStart) + operationParameterStart.length(); + int indexOfOperationEnd = originalMessage.indexOf("\",\"", indexOfOperationStart); + String badOperation = originalMessage.substring(indexOfOperationStart, indexOfOperationEnd); + return t.getMessage().replace("null", badOperation); } - private Mono invokePolicyAgent(DmaapRequestMessage dmaapRequestMessage) { + private Mono> invokePolicyAgent(DmaapRequestMessage dmaapRequestMessage) { DmaapRequestMessage.Operation operation = dmaapRequestMessage.operation(); - Mono result = null; String uri = dmaapRequestMessage.url(); + if (operation == Operation.DELETE) { - result = agentClient.delete(uri); + return agentClient.deleteForEntity(uri); } else if (operation == Operation.GET) { - result = agentClient.get(uri); + return agentClient.getForEntity(uri); } else if (operation == Operation.PUT) { - result = agentClient.put(uri, dmaapRequestMessage.payload()); + return agentClient.putForEntity(uri, payload(dmaapRequestMessage)); } else if (operation == Operation.POST) { - result = agentClient.post(uri, dmaapRequestMessage.payload()); + return agentClient.postForEntity(uri, payload(dmaapRequestMessage)); } else { - return Mono.error(new Exception("Not implemented operation: " + operation)); + return Mono.error(new ServiceException("Not implemented operation: " + operation)); + } + } + + private String payload(DmaapRequestMessage message) { + Optional payload = message.payload(); + if (payload.isPresent()) { + return gson.toJson(payload.get()); + } else { + logger.warn("Expected payload in message from DMAAP: {}", message); + return ""; } - return result; } private Mono sendDmaapResponse(String response, DmaapRequestMessage dmaapRequestMessage, HttpStatus status) { - return getDmaapResponseMessage(dmaapRequestMessage, response, status) // - .flatMap(body -> sendToDmaap(body)) // - .onErrorResume(t -> handleResponseCallError(t, dmaapRequestMessage)); + return createDmaapResponseMessage(dmaapRequestMessage, response, status) // + .flatMap(this::sendToDmaap) // + .onErrorResume(this::handleResponseCallError); } private Mono sendToDmaap(String body) { try { + logger.debug("sendToDmaap: {} ", body); dmaapClient.send(body); + dmaapClient.sendBatchWithResponse(); return Mono.just("OK"); } catch (IOException e) { return Mono.error(e); } } - private Mono handleResponseCallError(Throwable t, DmaapRequestMessage dmaapRequestMessage) { - logger.debug("Failed to respond: " + t.getMessage()); + private Mono handleResponseCallError(Throwable t) { + logger.debug("Failed to send response to DMaaP: {}", t.getMessage()); return Mono.empty(); } - private Mono getDmaapResponseMessage(DmaapRequestMessage dmaapRequestMessage, String response, + private Mono createDmaapResponseMessage(DmaapRequestMessage dmaapRequestMessage, String response, HttpStatus status) { DmaapResponseMessage dmaapResponseMessage = ImmutableDmaapResponseMessage.builder() // .status(status.toString()) // - .message(response) // + .message(response == null ? "" : response) // .type("response") // - .correlationId(dmaapRequestMessage.correlationId()) // - .originatorId(dmaapRequestMessage.originatorId()) // - .requestId(dmaapRequestMessage.requestId()) // - .timestamp(dmaapRequestMessage.timestamp()) // + .correlationId(dmaapRequestMessage.correlationId() == null ? "" : dmaapRequestMessage.correlationId()) // + .originatorId(dmaapRequestMessage.originatorId() == null ? "" : dmaapRequestMessage.originatorId()) // + .requestId(dmaapRequestMessage.requestId() == null ? "" : dmaapRequestMessage.requestId()) // + .timestamp(dmaapRequestMessage.timestamp() == null ? "" : dmaapRequestMessage.timestamp()) // .build(); String str = gson.toJson(dmaapResponseMessage); - return Mono.just(str); } - }