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
|
/*
* Copyright (C) 2021 Samsung Electronics
* 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
*/
package org.onap.rapp.datacollector.service;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.onap.rapp.datacollector.entity.ves.AdditionalMeasurementValues;
import org.onap.rapp.datacollector.entity.ves.AdditionalMeasurements;
import org.onap.rapp.datacollector.entity.ves.Event;
import org.onap.rapp.datacollector.service.configuration.DmaapRestReaderConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;
@Service
public class VesRetrievalService implements DmaapRestReader {
private static final Logger logger = LoggerFactory.getLogger(VesRetrievalService.class);
public static final String UE_FIELD_NAME = "trafficModel";
private final RestTemplate restTemplate;
private final DmaapRestReaderConfiguration config;
private final ParserFactory parser;
private final VesPersister persister;
private final UEHolder ueHolder;
@Autowired
public VesRetrievalService(RestTemplate restTemplate, ParserFactory parser, VesPersister persister,
DmaapRestReaderConfiguration configuration, UEHolder ueHolder) {
this.restTemplate = restTemplate;
this.parser = parser;
this.persister = persister;
this.config = configuration;
this.ueHolder = ueHolder;
}
@Override
public Collection<String> retrieveEvents() {
logger.info("Reaching from dmaap: {}", config.getMeasurementsTopicUrls());
return config.getMeasurementsTopicUrls().stream().flatMap(this::retrieveEventsFromTopic)
.collect(Collectors.toList());
}
private Stream<String> retrieveEventsFromTopic(String topic) {
try {
ResponseEntity<String[]> responseEntity = restTemplate.exchange(topic, HttpMethod.GET,
new HttpEntity<String>(createHeaders(config.getDmaapProperties().getUsername(),
config.getDmaapProperties().getPassword())), String[].class);
if (responseEntity.hasBody()) {
String[] events = responseEntity.getBody();
return Arrays.stream(events);
}
} catch (RestClientException ex) {
logger.error("Failed to reach to dmaap", ex);
}
return Arrays.stream(new String[0]);
}
private HttpHeaders createHeaders(String username, String password) {
HttpHeaders headers = new HttpHeaders();
headers.setBasicAuth(username, password);
return headers;
}
@Scheduled(fixedRate = 5000)
public void retrieveAndStoreVesEvents() {
retrieveEvents().stream().map(parser::getParsedEvents).forEach(this::saveAllEvents);
}
private void saveAllEvents(List<Event> events) {
saveUesOfVes(events);
persister.persistAll(removeTrafficModelFromEvents(events));
}
private void saveUesOfVes(List<Event> events) {
Set<String> uesOfVes = Optional.ofNullable(events).orElse(Collections.emptyList()).stream().flatMap(event -> getUserEquipmentData(event).stream())
.collect(Collectors.toSet());
uesOfVes.forEach(ueHolder::addUE);
}
private Set<String> getUserEquipmentData(Event event) {
Optional<AdditionalMeasurements> ues = event.getMeasurementFields().getAdditionalMeasurements()
.stream().filter(am -> am.getName().equals(UE_FIELD_NAME)).findAny();
return ues.map(additionalMeasurements -> additionalMeasurements.getValues().stream()
.map(AdditionalMeasurementValues::getParameterName)
.collect(Collectors.toSet())).orElse(Collections.emptySet());
}
private List<Event> removeTrafficModelFromEvents(List<Event> events) {
return events.stream().map(event -> {
event.getMeasurementFields().getAdditionalMeasurements()
.removeIf(additionalMeasurements -> UE_FIELD_NAME.equalsIgnoreCase(additionalMeasurements.getName()));
return event;
}).collect(Collectors.toList());
}
}
|
