X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=enrichment-coordinator-service%2Fsrc%2Fmain%2Fjava%2Forg%2Foransc%2Fenrichment%2Frepository%2FInfoTypeSubscriptions.java;h=60fe35f39452d73a8f1c852dde3c6ae18c148e58;hb=b0612ab177e14ffa133aae2538aa504d5cc10e99;hp=bdf0e8e4ca884b6576b80459cd841732d5b1e615;hpb=383f47b1ae6e767d0bd8334f4f45e4cf83bd7c32;p=nonrtric.git diff --git a/enrichment-coordinator-service/src/main/java/org/oransc/enrichment/repository/InfoTypeSubscriptions.java b/enrichment-coordinator-service/src/main/java/org/oransc/enrichment/repository/InfoTypeSubscriptions.java index bdf0e8e4..60fe35f3 100644 --- a/enrichment-coordinator-service/src/main/java/org/oransc/enrichment/repository/InfoTypeSubscriptions.java +++ b/enrichment-coordinator-service/src/main/java/org/oransc/enrichment/repository/InfoTypeSubscriptions.java @@ -20,34 +20,56 @@ package org.oransc.enrichment.repository; +import com.google.gson.Gson; +import com.google.gson.GsonBuilder; + +import java.io.File; +import java.io.FileOutputStream; +import java.io.IOException; +import java.io.PrintStream; import java.lang.invoke.MethodHandles; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.time.Duration; import java.util.Collection; import java.util.HashMap; import java.util.Map; import java.util.Vector; +import java.util.function.Function; import lombok.Builder; import lombok.Getter; +import org.oransc.enrichment.configuration.ApplicationConfig; import org.oransc.enrichment.exceptions.ServiceException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.springframework.stereotype.Component; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Configuration; +import org.springframework.util.FileSystemUtils; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.util.retry.Retry; /** * Subscriptions of callbacks for type registrations */ @SuppressWarnings("squid:S2629") // Invoke method(s) only conditionally -@Component +@Configuration public class InfoTypeSubscriptions { private final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); private final Map allSubscriptions = new HashMap<>(); private final MultiMap subscriptionsByOwner = new MultiMap<>(); + private final Gson gson = new GsonBuilder().create(); + private final ApplicationConfig config; + private final Map callbackHandlers = new HashMap<>(); - public interface Callbacks { - void notifyTypeRegistered(InfoType type, SubscriptionInfo subscriptionInfo); + public interface ConsumerCallbackHandler { + Mono notifyTypeRegistered(InfoType type, SubscriptionInfo subscriptionInfo); - void notifyTypeRemoved(InfoType type, SubscriptionInfo subscriptionInfo); + Mono notifyTypeRemoved(InfoType type, SubscriptionInfo subscriptionInfo); } @Builder @@ -59,12 +81,27 @@ public class InfoTypeSubscriptions { private String owner; - private Callbacks callback; + private String apiVersion; + } + + public InfoTypeSubscriptions(@Autowired ApplicationConfig config) { + this.config = config; + + try { + this.restoreFromDatabase(); + } catch (IOException e) { + logger.error("Could not restore info type subscriptions from database {}", this.getDatabaseDirectory()); + } + } + + public void registerCallbackhandler(ConsumerCallbackHandler handler, String apiVersion) { + callbackHandlers.put(apiVersion, handler); } public synchronized void put(SubscriptionInfo subscription) { allSubscriptions.put(subscription.getId(), subscription); subscriptionsByOwner.put(subscription.owner, subscription.id, subscription); + storeInFile(subscription); logger.debug("Added type status subscription {}", subscription.id); } @@ -105,18 +142,26 @@ public class InfoTypeSubscriptions { public synchronized void clear() { allSubscriptions.clear(); subscriptionsByOwner.clear(); + clearDatabase(); } public void remove(SubscriptionInfo subscription) { allSubscriptions.remove(subscription.getId()); subscriptionsByOwner.remove(subscription.owner, subscription.id); + + try { + Files.delete(getPath(subscription)); + } catch (Exception e) { + logger.debug("Could not delete subscription from database: {}", e.getMessage()); + } + logger.debug("Removed type status subscription {}", subscription.id); } /** * returns all subscriptions for an owner. The colllection can contain 0..n * subscriptions. - * + * * @param owner * @return */ @@ -125,11 +170,113 @@ public class InfoTypeSubscriptions { } public synchronized void notifyTypeRegistered(InfoType type) { - allSubscriptions.forEach((id, subscription) -> subscription.callback.notifyTypeRegistered(type, subscription)); + notifyAllSubscribers( + subscription -> getCallbacksHandler(subscription.apiVersion).notifyTypeRegistered(type, subscription)); } public synchronized void notifyTypeRemoved(InfoType type) { - allSubscriptions.forEach((id, subscription) -> subscription.callback.notifyTypeRemoved(type, subscription)); + notifyAllSubscribers( + subscription -> getCallbacksHandler(subscription.apiVersion).notifyTypeRemoved(type, subscription)); + } + + private ConsumerCallbackHandler getCallbacksHandler(String apiVersion) { + ConsumerCallbackHandler callbackHandler = this.callbackHandlers.get(apiVersion); + if (callbackHandler != null) { + return callbackHandler; + } else { + return new ConsumerCallbackHandler() { + @Override + public Mono notifyTypeRegistered(InfoType type, SubscriptionInfo subscriptionInfo) { + return error(); + } + + @Override + public Mono notifyTypeRemoved(InfoType type, SubscriptionInfo subscriptionInfo) { + return error(); + } + + private Mono error() { + return Mono.error(new ServiceException( + "No notifyTypeRegistered handler found for interface version " + apiVersion)); + } + }; + } + } + + private synchronized void notifyAllSubscribers(Function> notifyFunc) { + final int MAX_CONCURRENCY = 5; + Flux.fromIterable(allSubscriptions.values()) // + .flatMap(subscription -> notifySubscriber(notifyFunc, subscription), MAX_CONCURRENCY) // + .subscribe(); + } + + /** + * Invoking one consumer. If the call fails after retries, the subscription is + * removed. + * + * @param notifyFunc + * @param subscriptionInfo + * @return + */ + private Mono notifySubscriber(Function> notifyFunc, + SubscriptionInfo subscriptionInfo) { + Retry retrySpec = Retry.backoff(3, Duration.ofSeconds(1)); + return Mono.just(1) // + .flatMap(notUsed -> notifyFunc.apply(subscriptionInfo)) // + .retryWhen(retrySpec) // + .onErrorResume(throwable -> { + logger.warn("Consumer callback failed {}, removing subscription {}", throwable.getMessage(), + subscriptionInfo.id); + this.remove(subscriptionInfo); + return Mono.empty(); + }); // + } + + private void clearDatabase() { + try { + FileSystemUtils.deleteRecursively(Path.of(getDatabaseDirectory())); + Files.createDirectories(Paths.get(getDatabaseDirectory())); + } catch (IOException e) { + logger.warn("Could not delete database : {}", e.getMessage()); + } + } + + private void storeInFile(SubscriptionInfo subscription) { + try { + try (PrintStream out = new PrintStream(new FileOutputStream(getFile(subscription)))) { + String json = gson.toJson(subscription); + out.print(json); + } + } catch (Exception e) { + logger.warn("Could not save subscription: {} {}", subscription.getId(), e.getMessage()); + } + } + + public synchronized void restoreFromDatabase() throws IOException { + Files.createDirectories(Paths.get(getDatabaseDirectory())); + File dbDir = new File(getDatabaseDirectory()); + + for (File file : dbDir.listFiles()) { + String json = Files.readString(file.toPath()); + SubscriptionInfo subscription = gson.fromJson(json, SubscriptionInfo.class); + this.allSubscriptions.put(subscription.getId(), subscription); + } + } + + private File getFile(SubscriptionInfo subscription) { + return getPath(subscription).toFile(); + } + + private Path getPath(SubscriptionInfo subscription) { + return getPath(subscription.getId()); + } + + private Path getPath(String subscriptionId) { + return Path.of(getDatabaseDirectory(), subscriptionId); + } + + private String getDatabaseDirectory() { + return config.getVardataDirectory() + "/database/infotypesubscriptions"; } }