1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
/*-
* ============LICENSE_START=======================================================
* PROJECT
* ================================================================================
* Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
* ============LICENSE_END=========================================================
*/
package org.onap.dcae.commonFunction;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;
public class DmaapPropertyReader {
private static final Logger log = LoggerFactory.getLogger(DmaapPropertyReader.class);
private static final String CAMBRIA_TOPIC_KEY = "cambria.topic";
private static final String CAMBRIA_HOSTS_KEY = "cambria.hosts";
private static final String CAMBRIA_URL_KEY = "cambria.url";
private static final List<String> LEGACY_CHANNEL_MANDATORY_PARAMS = Lists.newArrayList(CAMBRIA_TOPIC_KEY, CAMBRIA_HOSTS_KEY, CAMBRIA_URL_KEY, "basicAuthPassword", "basicAuthUsername");
private static DmaapPropertyReader instance = null;
private final Map<String, String> dmaapProperties;
public DmaapPropertyReader(String cambriaConfigFilePath) {
this.dmaapProperties = DmaapPropertyReader.getProcessedDmaapProperties(cambriaConfigFilePath);
}
public Map<String, String> getDmaapProperties() {
return dmaapProperties;
}
public static synchronized DmaapPropertyReader getInstance(String channelConfig) {
if (instance == null) {
instance = new DmaapPropertyReader(channelConfig);
}
return instance;
}
public String getKeyValue(String hashKey) {
return dmaapProperties.get(hashKey);
}
private static Map<String, String> getProcessedDmaapProperties(String configFilePath) {
Map<String, String> transformedDmaapProperties = new HashMap<>();
try {
AnyNode root = AnyNode.parse(configFilePath);
if (isInLegacyFormat(root)) {
transformedDmaapProperties = getLegacyDmaapPropertiesWithChannels(root.get("channels"));
} else {
transformedDmaapProperties = getDmaapPropertiesWithInfoData(root);
}
} catch (IOException e) {
e.printStackTrace();
}
return transformedDmaapProperties;
}
private static boolean isInLegacyFormat(AnyNode root) {
return root.hasKey("channels");
}
private static Map<String, String> getLegacyDmaapPropertiesWithChannels(AnyNode channelsNode) {
return channelsNode.asList().stream()
.map(DmaapPropertyReader::getTransformedMandatoryChannelProperties)
.flatMap(m -> m.entrySet().stream())
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
}
private static Map<String, String> getTransformedMandatoryChannelProperties(AnyNode channel) {
String prefix = channel.get("name").asString() + ".";
return channel.asMap().entrySet().stream().filter(el -> LEGACY_CHANNEL_MANDATORY_PARAMS.contains(el.getKey()) && !Objects.equals(el.getKey(), "name"))
.collect(Collectors.toMap(k -> prefix + k.getKey(), v -> v.getValue().asString().replace("\"", "")));
}
private static Map<String, String> getDmaapPropertiesWithInfoData(AnyNode root) {
return root.asMap().entrySet().stream()
.map(DmaapPropertyReader::getTransformedMandatoryInfoProperties).flatMap(m -> m.entrySet().stream())
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
}
private static Map<String, String> getTransformedMandatoryInfoProperties(Map.Entry<String, AnyNode> el) {
String prefix = el.getKey() + ".";
AnyNode val = el.getValue();
Map<String, String> map = Maps.newHashMap();
map.put(prefix + "basicAuthUsername", val.getAsOptional("aaf_username").orElse(AnyNode.nullValue()).asString().replace("\"", ""));
map.put(prefix + "basicAuthPassword", val.getAsOptional("aaf_password").orElse(AnyNode.nullValue()).asString().replace("\"", ""));
map.putAll(getParamsFromDmaapInfoTopicUrl(prefix, val.get("dmaap_info").get("topic_url").asString().replace("\"", "")));
return map;
}
/***
* Dmaap url structure pub - https://<dmaaphostname>:<port>/events/
* <namespace>.<dmaapcluster>.<topic>, sub - https://<dmaaphostname>:
* <port>/events/<namespace>.<dmaapcluster>.<topic>/G1/u1";
*
* Onap url structure pub - http://<dmaaphostname>:<port>/<unauthenticated>.
* <topic>,
*/
private static Map<String, String> getParamsFromDmaapInfoTopicUrl(String keyPrefix, String topicUrl) {
Map<String, String> topicUrlParts = Maps.newHashMap();
try {
URL url = new URL(topicUrl);
topicUrlParts.put(keyPrefix + CAMBRIA_URL_KEY, url.getAuthority());
String[] pathParts = url.getPath().split("/");
if (pathParts.length > 2 && "events".equals(pathParts[1])) {
// DCAE internal dmaap topic convention
topicUrlParts.put(keyPrefix + CAMBRIA_TOPIC_KEY, pathParts[2]);
} else {
// ONAP dmaap topic convention
topicUrlParts.put(keyPrefix + CAMBRIA_TOPIC_KEY, pathParts[2]);
topicUrlParts.put(keyPrefix + CAMBRIA_HOSTS_KEY, url.getHost());
}
} catch (MalformedURLException e) {
log.error("Invalid URL found under topic_url key!", e);
e.printStackTrace();
}
return topicUrlParts;
}
}
|
