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 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<String, SubscriptionInfo> allSubscriptions = new HashMap<>();
private final MultiMap<SubscriptionInfo> subscriptionsByOwner = new MultiMap<>();
+ private final Gson gson = new GsonBuilder().create();
+ private final ApplicationConfig config;
+ private final Map<String, ConsumerCallbackHandler> callbackHandlers = new HashMap<>();
- public interface Callbacks {
+ public interface ConsumerCallbackHandler {
Mono<String> notifyTypeRegistered(InfoType type, SubscriptionInfo subscriptionInfo);
Mono<String> notifyTypeRemoved(InfoType type, SubscriptionInfo subscriptionInfo);
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);
}
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);
}
}
public synchronized void notifyTypeRegistered(InfoType type) {
- notifyAllSubscribers(subscription -> subscription.callback.notifyTypeRegistered(type, subscription));
+ notifyAllSubscribers(
+ subscription -> getCallbacksHandler(subscription.apiVersion).notifyTypeRegistered(type, subscription));
}
public synchronized void notifyTypeRemoved(InfoType type) {
- notifyAllSubscribers(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<String> notifyTypeRegistered(InfoType type, SubscriptionInfo subscriptionInfo) {
+ return error();
+ }
+
+ @Override
+ public Mono<String> notifyTypeRemoved(InfoType type, SubscriptionInfo subscriptionInfo) {
+ return error();
+ }
+
+ private Mono<String> error() {
+ return Mono.error(new ServiceException(
+ "No notifyTypeRegistered handler found for interface version " + apiVersion));
+ }
+ };
+ }
}
private synchronized void notifyAllSubscribers(Function<? super SubscriptionInfo, Mono<String>> notifyFunc) {
- final int CONCURRENCY = 5;
+ final int MAX_CONCURRENCY = 5;
Flux.fromIterable(allSubscriptions.values()) //
- .flatMap(notifyFunc::apply, CONCURRENCY) //
- .onErrorResume(throwable -> {
- logger.warn("Post failed for consumer callback {}", throwable.getMessage());
- return Flux.empty();
- }) //
+ .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<String> notifySubscriber(Function<? super SubscriptionInfo, Mono<String>> 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";
+ }
+
}
Code Review / nonrtric.git / blobdiff