import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
-import java.lang.invoke.MethodHandles;
-
import org.oransc.enrichment.clients.AsyncRestClient;
import org.oransc.enrichment.clients.AsyncRestClientFactory;
import org.oransc.enrichment.configuration.ApplicationConfig;
import org.oransc.enrichment.repository.InfoType;
import org.oransc.enrichment.repository.InfoTypeSubscriptions;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
+import reactor.core.publisher.Mono;
/**
* Callbacks to the Consumer. Notifies consumer according to the API (which this
@Component
public class ConsumerCallbacks implements InfoTypeSubscriptions.Callbacks {
- private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private static Gson gson = new GsonBuilder().create();
private final AsyncRestClient restClient;
}
@Override
- public void notifyTypeRegistered(InfoType type, InfoTypeSubscriptions.SubscriptionInfo subscriptionInfo) {
+ public Mono<String> notifyTypeRegistered(InfoType type, InfoTypeSubscriptions.SubscriptionInfo subscriptionInfo) {
ConsumerTypeRegistrationInfo info = new ConsumerTypeRegistrationInfo(type.getJobDataSchema(),
ConsumerTypeRegistrationInfo.ConsumerTypeStatusValues.REGISTERED, type.getId());
String body = gson.toJson(info);
- post(subscriptionInfo.getCallbackUrl(), body);
+ return restClient.post(subscriptionInfo.getCallbackUrl(), body);
}
@Override
- public void notifyTypeRemoved(InfoType type, InfoTypeSubscriptions.SubscriptionInfo subscriptionInfo) {
+ public Mono<String> notifyTypeRemoved(InfoType type, InfoTypeSubscriptions.SubscriptionInfo subscriptionInfo) {
ConsumerTypeRegistrationInfo info = new ConsumerTypeRegistrationInfo(type.getJobDataSchema(),
ConsumerTypeRegistrationInfo.ConsumerTypeStatusValues.DEREGISTERED, type.getId());
String body = gson.toJson(info);
- post(subscriptionInfo.getCallbackUrl(), body);
-
- }
-
- private void post(String url, String body) {
- restClient.post(url, body) //
- .subscribe(response -> logger.debug("Post OK {}", url), //
- throwable -> logger.warn("Post failed for consumer callback {} {}", url, body), null);
+ return restClient.post(subscriptionInfo.getCallbackUrl(), body);
}
}
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.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
/**
* Subscriptions of callbacks for type registrations
*/
private final MultiMap<SubscriptionInfo> subscriptionsByOwner = new MultiMap<>();
public interface Callbacks {
- void notifyTypeRegistered(InfoType type, SubscriptionInfo subscriptionInfo);
+ Mono<String> notifyTypeRegistered(InfoType type, SubscriptionInfo subscriptionInfo);
- void notifyTypeRemoved(InfoType type, SubscriptionInfo subscriptionInfo);
+ Mono<String> notifyTypeRemoved(InfoType type, SubscriptionInfo subscriptionInfo);
}
@Builder
/**
* returns all subscriptions for an owner. The colllection can contain 0..n
* subscriptions.
- *
+ *
* @param owner
* @return
*/
}
public synchronized void notifyTypeRegistered(InfoType type) {
- allSubscriptions.forEach((id, subscription) -> subscription.callback.notifyTypeRegistered(type, subscription));
+ notifyAllSubscribers(subscription -> subscription.callback.notifyTypeRegistered(type, subscription));
}
public synchronized void notifyTypeRemoved(InfoType type) {
- allSubscriptions.forEach((id, subscription) -> subscription.callback.notifyTypeRemoved(type, subscription));
+ notifyAllSubscribers(subscription -> subscription.callback.notifyTypeRemoved(type, subscription));
+ }
+
+ private synchronized void notifyAllSubscribers(Function<? super SubscriptionInfo, Mono<String>> notifyFunc) {
+ final int CONCURRENCY = 5;
+ Flux.fromIterable(allSubscriptions.values()) //
+ .flatMap(notifyFunc::apply, CONCURRENCY) //
+ .onErrorResume(throwable -> {
+ logger.warn("Post failed for consumer callback {}", throwable.getMessage());
+ return Flux.empty();
+ }) //
+ .subscribe();
}
}