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=226b54e9a2093612bd38491bfe3be3b6ff97cb49;hb=6a39814272307d0207222c9229b0d765ac062bf0;hp=2d963c35e5e24761174f249ad4e57e71571d87b2;hpb=0da7354bf7548829a7152dfa0276710dbbe00b7b;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 2d963c35..226b54e9 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,136 @@ package org.oransc.policyagent.dmaap; import com.google.gson.Gson; import com.google.gson.GsonBuilder; -import java.io.IOException; -import org.onap.dmaap.mr.client.MRBatchingPublisher; +import com.google.gson.JsonObject; + +import java.util.Optional; + 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; + .create(); // + private final AsyncRestClient dmaapClient; private final AsyncRestClient agentClient; - public DmaapMessageHandler(MRBatchingPublisher dmaapClient, ApplicationConfig applicationConfig, - AsyncRestClient agentClient) { + public DmaapMessageHandler(AsyncRestClient 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, 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, DmaapRequestMessage dmaapRequestMessage) { - logger.debug("Agent call failed: " + t.getMessage()); - return sendDmaapResponse(t.toString(), dmaapRequestMessage, HttpStatus.NOT_FOUND) // - .flatMap(s -> Mono.empty()); + private Mono> handleAgentCallError(Throwable error, + DmaapRequestMessage dmaapRequestMessage) { + logger.debug("Agent call failed: {}", error.getMessage()); + HttpStatus status = HttpStatus.INTERNAL_SERVER_ERROR; + String errorMessage = error.getMessage(); + if (error instanceof WebClientResponseException) { + WebClientResponseException exception = (WebClientResponseException) error; + status = exception.getStatusCode(); + errorMessage = exception.getResponseBodyAsString(); + } else if (error instanceof ServiceException) { + status = HttpStatus.BAD_REQUEST; + errorMessage = error.getMessage(); + } else if (!(error instanceof WebClientException)) { + logger.warn("Unexpected exception ", error); + } + return sendDmaapResponse(errorMessage, dmaapRequestMessage, status) // + .flatMap(notUsed -> Mono.empty()); } - 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 ServiceException("Not implemented operation: " + operation)); + } + } + + private String payload(DmaapRequestMessage message) { + Optional payload = message.payload(); + if (payload.isPresent()) { + return gson.toJson(payload.get()); } else { - return Mono.error(new Exception("Not implemented operation: " + operation)); + 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)); + HttpStatus status) { + 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); - } + logger.debug("sendToDmaap: {} ", body); + return dmaapClient.post("", "[" + body + "]"); } - 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, - HttpStatus status) { + private Mono createDmaapResponseMessage(DmaapRequestMessage dmaapRequestMessage, String response, + HttpStatus status) { DmaapResponseMessage dmaapResponseMessage = ImmutableDmaapResponseMessage.builder() // - .status(status.toString()) // - .message(response) // - .type("response") // - .correlationId(dmaapRequestMessage.correlationId()) // - .originatorId(dmaapRequestMessage.originatorId()) // - .requestId(dmaapRequestMessage.requestId()) // - .timestamp(dmaapRequestMessage.timestamp()) // - .build(); + .status(status.toString()) // + .message(response == null ? "" : response) // + .type("response") // + .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); } - }