2 * ========================LICENSE_START=================================
5 * Copyright (C) 2023 Nordix Foundation
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
11 * http://www.apache.org/licenses/LICENSE-2.0
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===================================
21 package org.oran.pmproducer.repository;
23 import java.util.ArrayList;
24 import java.util.Collection;
25 import java.util.HashMap;
26 import java.util.List;
28 import java.util.Vector;
32 import org.oran.pmproducer.clients.AsyncRestClientFactory;
33 import org.oran.pmproducer.clients.SecurityContext;
34 import org.oran.pmproducer.configuration.ApplicationConfig;
35 import org.oran.pmproducer.exceptions.ServiceException;
36 import org.oran.pmproducer.filter.FilterFactory;
37 import org.oran.pmproducer.filter.FilteredData;
38 import org.oran.pmproducer.filter.PmReportFilter;
39 import org.oran.pmproducer.repository.Job.Parameters;
40 import org.oran.pmproducer.repository.Job.Parameters.KafkaDeliveryInfo;
41 import org.oran.pmproducer.tasks.TopicListener.DataFromTopic;
42 import org.slf4j.Logger;
43 import org.slf4j.LoggerFactory;
44 import org.springframework.beans.factory.annotation.Autowired;
45 import org.springframework.http.HttpStatus;
46 import org.springframework.stereotype.Component;
50 public interface Observer {
51 void onJobbGroupAdded(JobGroup jobGroup);
53 void onJobGroupRemoved(JobGroup jobGroup);
56 public static class JobGroup {
58 private final KafkaDeliveryInfo deliveryInfo;
60 private Map<String, Job> jobs = new HashMap<>();
63 private PmReportFilter filter;
66 private final InfoType type;
68 public JobGroup(InfoType type, KafkaDeliveryInfo deliveryInfo) {
69 this.deliveryInfo = deliveryInfo;
73 public synchronized void add(Job job) {
74 this.jobs.put(job.getId(), job);
75 this.filter = createFilter();
78 public synchronized void remove(Job job) {
79 this.jobs.remove(job.getId());
80 if (!this.jobs.isEmpty()) {
81 this.filter = createFilter();
85 public boolean isEmpty() {
86 return jobs.isEmpty();
89 public FilteredData filter(DataFromTopic data) {
90 return filter.filter(data);
93 public Job getAJob() {
94 if (this.jobs.isEmpty()) {
97 return this.jobs.values().iterator().next();
100 private PmReportFilter createFilter() {
101 Collection<PmReportFilter> filterData = new ArrayList<>();
102 this.jobs.forEach((key, value) -> filterData.add(value.getFilter()));
103 return FilterFactory.createAggregateFilter(filterData);
106 public String getId() {
107 return deliveryInfo.getTopic();
110 public Iterable<Job> getJobs() {
111 return this.jobs.values();
115 private static final Logger logger = LoggerFactory.getLogger(Jobs.class);
117 private Map<String, Job> allJobs = new HashMap<>();
118 private MultiMap<Job> jobsByType = new MultiMap<>();
119 private Map<String, JobGroup> jobGroups = new HashMap<>(); // Key is Topic
120 private final AsyncRestClientFactory restclientFactory;
121 private final List<Observer> observers = new ArrayList<>();
122 private final ApplicationConfig appConfig;
124 public Jobs(@Autowired ApplicationConfig applicationConfig, @Autowired SecurityContext securityContext,
125 @Autowired ApplicationConfig appConfig) {
126 restclientFactory = new AsyncRestClientFactory(applicationConfig.getWebClientConfig(), securityContext);
127 this.appConfig = appConfig;
130 public synchronized Job getJob(String id) throws ServiceException {
131 Job job = allJobs.get(id);
133 throw new ServiceException("Could not find job: " + id, HttpStatus.NOT_FOUND);
138 public synchronized Job get(String id) {
139 return allJobs.get(id);
142 public void addJob(String id, InfoType type, String owner, String lastUpdated, Parameters parameters)
143 throws ServiceException {
145 Job job = new Job(id, type, owner, lastUpdated, parameters, this.appConfig);
149 public void addObserver(Observer obs) {
150 synchronized (observers) {
151 this.observers.add(obs);
155 private String jobGroupId(Job job) {
156 return job.getParameters().getDeliveryInfo().getTopic();
159 private synchronized void put(Job job) {
160 logger.debug("Put job: {}", job.getId());
163 allJobs.put(job.getId(), job);
164 jobsByType.put(job.getType().getId(), job.getId(), job);
166 String jobGroupId = jobGroupId(job);
167 if (!this.jobGroups.containsKey(jobGroupId)) {
168 final JobGroup group = new JobGroup(job.getType(), job.getParameters().getDeliveryInfo());
169 this.jobGroups.put(jobGroupId, group);
171 this.observers.forEach(obs -> obs.onJobbGroupAdded(group));
173 JobGroup group = this.jobGroups.get(jobGroupId);
178 public synchronized Iterable<Job> getAll() {
179 return new Vector<>(allJobs.values());
182 public synchronized Job remove(String id) {
183 Job job = allJobs.get(id);
190 public void remove(Job job) {
191 String groupId = jobGroupId(job);
192 JobGroup group = this.jobGroups.get(groupId);
193 synchronized (this) {
194 this.allJobs.remove(job.getId());
195 jobsByType.remove(job.getType().getId(), job.getId());
197 if (group.isEmpty()) {
198 this.jobGroups.remove(groupId);
202 if (group.isEmpty()) {
203 this.observers.forEach(obs -> obs.onJobGroupRemoved(group));
207 public synchronized int size() {
208 return allJobs.size();
211 public synchronized Collection<Job> getJobsForType(InfoType type) {
212 return jobsByType.get(type.getId());
215 public void clear() {
217 this.jobGroups.forEach((id, group) -> this.observers.forEach(obs -> obs.onJobGroupRemoved(group)));
219 synchronized (this) {