Running Dmaap consumer in a seprate thread
[nonrtric.git] / policy-agent / src / main / java / org / oransc / policyagent / clients / OscA1Client.java
1 /*-
2  * ========================LICENSE_START=================================
3  * O-RAN-SC
4  * %%
5  * Copyright (C) 2019 Nordix Foundation
6  * %%
7  * Licensed under the Apache License, Version 2.0 (the "License");
8  * you may not use this file except in compliance with the License.
9  * You may obtain a copy of the License at
10  *
11  *      http://www.apache.org/licenses/LICENSE-2.0
12  *
13  * Unless required by applicable law or agreed to in writing, software
14  * distributed under the License is distributed on an "AS IS" BASIS,
15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16  * See the License for the specific language governing permissions and
17  * limitations under the License.
18  * ========================LICENSE_END===================================
19  */
20
21 package org.oransc.policyagent.clients;
22
23 import java.lang.invoke.MethodHandles;
24 import java.util.ArrayList;
25 import java.util.List;
26
27 import org.json.JSONArray;
28 import org.json.JSONException;
29 import org.json.JSONObject;
30 import org.oransc.policyagent.configuration.RicConfig;
31 import org.oransc.policyagent.repository.Policy;
32 import org.slf4j.Logger;
33 import org.slf4j.LoggerFactory;
34
35 import reactor.core.publisher.Flux;
36 import reactor.core.publisher.Mono;
37
38 public class OscA1Client implements A1Client {
39     private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
40
41     private final AsyncRestClient restClient;
42
43     public OscA1Client(RicConfig ricConfig) {
44         String baseUrl = ricConfig.baseUrl() + "/a1-p";
45         this.restClient = new AsyncRestClient(baseUrl);
46         logger.debug("OscA1Client for ric: {}", ricConfig.name());
47     }
48
49     @Override
50     public Mono<List<String>> getPolicyTypeIdentities() {
51         return restClient.get("/policytypes") //
52             .flatMap(this::parseJsonArrayOfString);
53     }
54
55     @Override
56     public Mono<List<String>> getPolicyIdentities() {
57         return getPolicyTypeIdentities() //
58             .flatMapMany(types -> Flux.fromIterable(types)) //
59             .flatMap(type -> getPolicyIdentities(type)) //
60             .flatMap(policyIds -> Flux.fromIterable(policyIds)) //
61             .collectList();
62     }
63
64     private Mono<List<String>> getPolicyIdentities(String typeId) {
65         return restClient.get("/policytypes/" + typeId + "/policies") //
66             .flatMap(this::parseJsonArrayOfString);
67     }
68
69     @Override
70     public Mono<String> getPolicyTypeSchema(String policyTypeId) {
71         return restClient.get("/policytypes/" + policyTypeId) //
72             .flatMap(response -> getCreateSchema(response, policyTypeId));
73     }
74
75     private Mono<String> getCreateSchema(String policyTypeResponse, String policyTypeId) {
76         try {
77             JSONObject obj = new JSONObject(policyTypeResponse);
78             JSONObject schemaObj = obj.getJSONObject("create_schema");
79             schemaObj.put("title", policyTypeId);
80             return Mono.just(schemaObj.toString());
81         } catch (Exception e) {
82             logger.error("Unexcpected response for policy type: {}", policyTypeResponse, e);
83             return Mono.error(e);
84         }
85     }
86
87     @Override
88     public Mono<String> putPolicy(Policy policy) {
89         return restClient.put("/policytypes/" + policy.type().name() + "/policies/" + policy.id(), policy.json());
90     }
91
92     @Override
93     public Mono<String> deletePolicy(Policy policy) {
94         return deletePolicy(policy.type().name(), policy.id());
95     }
96
97     private Mono<String> deletePolicy(String typeId, String policyId) {
98         return restClient.delete("/policytypes/" + typeId + "/policies/" + policyId);
99     }
100
101     @Override
102     public Mono<A1ProtocolType> getProtocolVersion() {
103         return restClient.get("/healthcheck") //
104             .flatMap(resp -> Mono.just(A1ProtocolType.OSC_V1));
105     }
106
107     @Override
108     public Flux<String> deleteAllPolicies() {
109         return getPolicyTypeIdentities() //
110             .flatMapMany(types -> Flux.fromIterable(types)) //
111             .flatMap(typeId -> deletePoliciesForType(typeId)); //
112     }
113
114     private Flux<String> deletePoliciesForType(String typeId) {
115         return getPolicyIdentities(typeId) //
116             .flatMapMany(policyIds -> Flux.fromIterable(policyIds)) //
117             .flatMap(policyId -> deletePolicy(typeId, policyId)); //
118     }
119
120     private Mono<List<String>> parseJsonArrayOfString(String inputString) {
121         try {
122             List<String> arrayList = new ArrayList<>();
123             JSONArray jsonArray = new JSONArray(inputString);
124             for (int i = 0; i < jsonArray.length(); i++) {
125                 arrayList.add(jsonArray.getString(i));
126             }
127             logger.debug("A1 client: received list = {}", arrayList);
128             return Mono.just(arrayList);
129         } catch (JSONException ex) { // invalid json
130             return Mono.error(ex);
131         }
132     }
133
134     @Override
135     public Mono<String> getPolicyStatus(Policy policy) {
136         // /a1-p/policytypes/{policy_type_id}/policies/{policy_instance_id}/status
137         return restClient.get("/policytypes/" + policy.type().name() + "/policies/" + policy.id() + "/status");
138
139     }
140 }