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
* class adapts to)
*/
-@SuppressWarnings("java:S3457") // No need to call "toString()" method as formatting and string ..
@Component
-public class ConsumerCallbacks implements InfoTypeSubscriptions.Callbacks {
+public class ConsumerCallbacks implements InfoTypeSubscriptions.ConsumerCallbackHandler {
- private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private static Gson gson = new GsonBuilder().create();
private final AsyncRestClient restClient;
- public ConsumerCallbacks(@Autowired ApplicationConfig config) {
+ public static final String API_VERSION = "version_1";
+
+ public ConsumerCallbacks(@Autowired ApplicationConfig config,
+ @Autowired InfoTypeSubscriptions infoTypeSubscriptions) {
AsyncRestClientFactory restClientFactory = new AsyncRestClientFactory(config.getWebClientConfig());
this.restClient = restClientFactory.createRestClientNoHttpProxy("");
+ infoTypeSubscriptions.registerCallbackhandler(this, API_VERSION);
}
@Override
- public void 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);
-
+ public Mono<String> notifyTypeRegistered(InfoType type, InfoTypeSubscriptions.SubscriptionInfo subscriptionInfo) {
+ String body = body(type, ConsumerTypeRegistrationInfo.ConsumerTypeStatusValues.REGISTERED);
+ return restClient.post(subscriptionInfo.getCallbackUrl(), body);
}
@Override
- public void 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);
-
+ public Mono<String> notifyTypeRemoved(InfoType type, InfoTypeSubscriptions.SubscriptionInfo subscriptionInfo) {
+ String body = body(type, ConsumerTypeRegistrationInfo.ConsumerTypeStatusValues.DEREGISTERED);
+ return restClient.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);
+ private String body(InfoType type, ConsumerTypeRegistrationInfo.ConsumerTypeStatusValues status) {
+ ConsumerTypeRegistrationInfo info =
+ new ConsumerTypeRegistrationInfo(type.getJobDataSchema(), status, type.getId());
+ return gson.toJson(info);
}
}