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.configuration.ApplicationConfig;
33 import org.oran.pmproducer.exceptions.ServiceException;
34 import org.oran.pmproducer.filter.FilterFactory;
35 import org.oran.pmproducer.filter.FilteredData;
36 import org.oran.pmproducer.filter.PmReportFilter;
37 import org.oran.pmproducer.oauth2.SecurityContext;
38 import org.oran.pmproducer.repository.Job.Parameters;
39 import org.oran.pmproducer.repository.Job.Parameters.KafkaDeliveryInfo;
40 import org.oran.pmproducer.tasks.TopicListener.DataFromTopic;
41 import org.slf4j.Logger;
42 import org.slf4j.LoggerFactory;
43 import org.springframework.beans.factory.annotation.Autowired;
44 import org.springframework.http.HttpStatus;
45 import org.springframework.stereotype.Component;
49 public interface Observer {
50 void onJobbGroupAdded(JobGroup jobGroup);
52 void onJobGroupRemoved(JobGroup jobGroup);
55 public static class JobGroup {
57 private final KafkaDeliveryInfo deliveryInfo;
59 private Map<String, Job> jobs = new HashMap<>();
62 private PmReportFilter filter;
65 private final InfoType type;
67 public JobGroup(InfoType type, KafkaDeliveryInfo deliveryInfo) {
68 this.deliveryInfo = deliveryInfo;
72 public synchronized void add(Job job) {
73 this.jobs.put(job.getId(), job);
74 this.filter = createFilter();
77 public synchronized void remove(Job job) {
78 this.jobs.remove(job.getId());
79 if (!this.jobs.isEmpty()) {
80 this.filter = createFilter();
84 public boolean isEmpty() {
85 return jobs.isEmpty();
88 public FilteredData filter(DataFromTopic data) {
89 return filter.filter(data);
92 public Job getAJob() {
93 if (this.jobs.isEmpty()) {
96 return this.jobs.values().iterator().next();
99 private PmReportFilter createFilter() {
100 Collection<PmReportFilter> filterData = new ArrayList<>();
101 this.jobs.forEach((key, value) -> filterData.add(value.getFilter()));
102 return FilterFactory.createAggregateFilter(filterData);
105 public String getId() {
106 return deliveryInfo.getTopic();
109 public Iterable<Job> getJobs() {
110 return this.jobs.values();
114 private static final Logger logger = LoggerFactory.getLogger(Jobs.class);
116 private Map<String, Job> allJobs = new HashMap<>();
117 private MultiMap<Job> jobsByType = new MultiMap<>();
118 private Map<String, JobGroup> jobGroups = new HashMap<>(); // Key is Topic
119 private final List<Observer> observers = new ArrayList<>();
120 private final ApplicationConfig appConfig;
122 public Jobs(@Autowired ApplicationConfig applicationConfig, @Autowired SecurityContext securityContext,
123 @Autowired ApplicationConfig appConfig) {
124 this.appConfig = appConfig;
127 public synchronized Job getJob(String id) throws ServiceException {
128 Job job = allJobs.get(id);
130 throw new ServiceException("Could not find job: " + id, HttpStatus.NOT_FOUND);
135 public synchronized Job get(String id) {
136 return allJobs.get(id);
139 public void addJob(String id, InfoType type, String owner, String lastUpdated, Parameters parameters)
140 throws ServiceException {
142 Job job = new Job(id, type, owner, lastUpdated, parameters, this.appConfig);
146 public void addObserver(Observer obs) {
147 synchronized (observers) {
148 this.observers.add(obs);
152 private String jobGroupId(Job job) {
153 return job.getParameters().getDeliveryInfo().getTopic();
156 private synchronized void put(Job job) {
157 logger.debug("Put job: {}", job.getId());
160 allJobs.put(job.getId(), job);
161 jobsByType.put(job.getType().getId(), job.getId(), job);
163 String jobGroupId = jobGroupId(job);
164 if (!this.jobGroups.containsKey(jobGroupId)) {
165 final JobGroup group = new JobGroup(job.getType(), job.getParameters().getDeliveryInfo());
166 this.jobGroups.put(jobGroupId, group);
168 this.observers.forEach(obs -> obs.onJobbGroupAdded(group));
170 JobGroup group = this.jobGroups.get(jobGroupId);
175 public synchronized Iterable<Job> getAll() {
176 return new Vector<>(allJobs.values());
179 public synchronized Job remove(String id) {
180 Job job = allJobs.get(id);
187 public void remove(Job job) {
188 String groupId = jobGroupId(job);
189 JobGroup group = this.jobGroups.get(groupId);
190 synchronized (this) {
191 this.allJobs.remove(job.getId());
192 jobsByType.remove(job.getType().getId(), job.getId());
194 if (group.isEmpty()) {
195 this.jobGroups.remove(groupId);
199 if (group.isEmpty()) {
200 this.observers.forEach(obs -> obs.onJobGroupRemoved(group));
204 public synchronized int size() {
205 return allJobs.size();
208 public synchronized Collection<Job> getJobsForType(InfoType type) {
209 return jobsByType.get(type.getId());
212 public void clear() {
214 this.jobGroups.forEach((id, group) -> this.observers.forEach(obs -> obs.onJobGroupRemoved(group)));
216 synchronized (this) {