pmproducer/src/main/java/org/oran/pmproducer/tasks/TopicListeners.java

   1 /*-
   2  * ========================LICENSE_START=================================
   3  * O-RAN-SC
   4  * %%
   5  * Copyright (C) 2023 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.oran.pmproducer.tasks;
  22
  23 import java.util.HashMap;
  24 import java.util.Map;
  25
  26 import lombok.Getter;
  27
  28 import org.oran.pmproducer.configuration.ApplicationConfig;
  29 import org.oran.pmproducer.repository.InfoType;
  30 import org.oran.pmproducer.repository.InfoTypes;
  31 import org.oran.pmproducer.repository.Jobs;
  32 import org.oran.pmproducer.repository.Jobs.JobGroup;
  33 import org.oran.pmproducer.repository.MultiMap;
  34 import org.slf4j.Logger;
  35 import org.slf4j.LoggerFactory;
  36 import org.springframework.beans.factory.annotation.Autowired;
  37 import org.springframework.scheduling.annotation.EnableScheduling;
  38 import org.springframework.stereotype.Component;
  39
  40 @SuppressWarnings("squid:S2629") // Invoke method(s) only conditionally
  41 @Component
  42 @EnableScheduling
  43 public class TopicListeners {
  44     private static final Logger logger = LoggerFactory.getLogger(TopicListeners.class);
  45
  46     @SuppressWarnings("java:S1700")
  47     @Getter
  48     private final Map<String, TopicListener> topicListeners = new HashMap<>(); // Key is typeId
  49
  50     @Getter
  51     private final MultiMap<JobDataDistributor> dataDistributors = new MultiMap<>(); // Key is typeId, jobId
  52
  53     private final ApplicationConfig appConfig;
  54
  55     public TopicListeners(@Autowired ApplicationConfig appConfig, @Autowired InfoTypes types, @Autowired Jobs jobs) {
  56         this.appConfig = appConfig;
  57
  58         for (InfoType type : types.getAll()) {
  59             TopicListener topicConsumer = new TopicListener(appConfig, type);
  60             topicListeners.put(type.getId(), topicConsumer);
  61         }
  62
  63         jobs.addObserver(new Jobs.Observer() {
  64             @Override
  65             public void onJobbGroupAdded(JobGroup jobGroup) {
  66                 addJob(jobGroup);
  67             }
  68
  69             @Override
  70             public void onJobGroupRemoved(JobGroup jobGroup) {
  71                 removeDistributor(jobGroup);
  72             }
  73         });
  74     }
  75
  76     public synchronized void addJob(JobGroup jobGroup) {
  77         removeDistributor(jobGroup);
  78         logger.debug("Job added {}", jobGroup.getId());
  79         addDistributor(jobGroup, dataDistributors, topicListeners);
  80     }
  81
  82     private JobDataDistributor createDistributor(JobGroup jobGroup) {
  83         return new JobDataDistributor(jobGroup, appConfig);
  84     }
  85
  86     private void addDistributor(JobGroup jobGroup, MultiMap<JobDataDistributor> distributors,
  87             Map<String, TopicListener> topicListeners) {
  88         TopicListener topicListener = topicListeners.get(jobGroup.getType().getId());
  89         JobDataDistributor distributor = createDistributor(jobGroup);
  90
  91         distributor.start(topicListener.getFlux());
  92
  93         distributors.put(jobGroup.getType().getId(), jobGroup.getId(), distributor);
  94     }
  95
  96     private synchronized void removeDistributor(JobGroup jobGroup) {
  97         removeDistributor(jobGroup, dataDistributors);
  98     }
  99
 100     private static void removeDistributor(JobGroup jobGroup, MultiMap<JobDataDistributor> distributors) {
 101         JobDataDistributor distributor = distributors.remove(jobGroup.getType().getId(), jobGroup.getId());
 102         if (distributor != null) {
 103             logger.debug("Job removed {}", jobGroup.getId());
 104             distributor.stop();
 105         }
 106     }
 107
 108 }