#104

Committed 01 Oct 2026 07:34AM UTC coverage: 4.489% (+0.7%) from 3.796%

Build # #104

Build Type

Committed by

Commit Message

Remove unused exception constructors, explain the handler defer

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Pull Request Pull Request #28: Add failure reasons, downstream error descriptions and configurable log output

Coverage Stats

218 of 379 new or added lines in 33 files covered. (57.52%)

9 existing lines in 6 files now uncovered.

1606 of 35773 relevant lines covered (4.49%)

0.04 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

60.98

/src/main/java/org/fiware/iam/tmforum/notification/NotificationSubscriber.java

package org.fiware.iam.tmforum.notification;

import io.micronaut.context.annotation.Context;
import io.micronaut.context.annotation.Requires;
import io.micronaut.context.annotation.Value;
import io.micronaut.http.HttpMethod;
import io.micronaut.http.HttpRequest;
import io.micronaut.http.HttpStatus;
import io.micronaut.http.client.exceptions.HttpClientResponseException;
import io.micronaut.http.client.netty.DefaultHttpClient;
import io.micronaut.runtime.event.annotation.EventListener;
import io.micronaut.runtime.server.event.ServerStartupEvent;
import io.micronaut.scheduling.TaskScheduler;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.fiware.iam.logging.DownstreamError;
import org.fiware.iam.configuration.GeneralProperties;
import org.fiware.iam.configuration.NotificationProperties;
import org.fiware.iam.tmforum.party.model.EventSubscriptionInputVO;
import org.fiware.iam.tmforum.party.model.EventSubscriptionVO;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.List;
import java.util.Optional;

/**
 * Responsible for subscribing to all configured tmforum events.
 */
@Requires(condition = NotificationProperties.NotificationCondition.class)
@Context
@RequiredArgsConstructor
@Slf4j
public class NotificationSubscriber {

    private static final String QUERY_TEMPLATE = "eventType=%s%s";
    private static final String LISTENER_ADDRESS_TEMPLATE = "%s/hub";
    private static final String LISTENER_PATH = "/listener/event";

    private final SubscriptionHealthIndicator subscriptionHealthIndicator;
    private final NotificationProperties notificationProperties;
    private final GeneralProperties generalProperties;
    private final DefaultHttpClient httpClient;
    private final TaskScheduler taskScheduler;

    @Value("${micronaut.server.port:8080}")
    private String servicePort;

    private static String removeTrailingSlash(String path) {
        if (path.endsWith("/")) {
            return path.substring(0, path.length() - 1);
        }
        return path;
    }

    @EventListener
    public void onApplicationEvent(ServerStartupEvent e) {
        notificationProperties.getEntities()
                .forEach(tmForumEntity ->
                        Optional.ofNullable(tmForumEntity.getEventTypes())
                                .orElse(List.of())
                                .forEach(eventType -> {
                                    subscriptionHealthIndicator.initiateSubscriptionInMap(tmForumEntity.getEntityType() + eventType.getValue());
                                    scheduleSubscription(notificationProperties.getSubscriptionInitialDelay(), tmForumEntity.getEntityType(), eventType.getValue(), tmForumEntity.getApiAddress());
                                }));
    }

    private void scheduleSubscription(long delaySeconds, String entityType, String eventType, String apiAddress) {
        taskScheduler.schedule(Duration.ofSeconds(delaySeconds), () -> createSubscription(entityType, eventType, apiAddress));
    }

    public void createSubscription(String entityType, String eventType, String apiAddress) {
        String callbackUrl = String.format("http://%s:%s%s%s", notificationProperties.getHost(), servicePort, removeTrailingSlash(generalProperties.getBasePath()), LISTENER_PATH);
        log.debug("Attempting to register subscription for {} {} events at {}", entityType, eventType, String.format(LISTENER_ADDRESS_TEMPLATE, apiAddress));

        EventSubscriptionInputVO subscription = new EventSubscriptionInputVO()
                .callback(callbackUrl)
                .query(String.format(QUERY_TEMPLATE, entityType, eventType));

        HttpRequest<?> request = HttpRequest.create(HttpMethod.POST, String.format(LISTENER_ADDRESS_TEMPLATE, apiAddress)).body(subscription);

        Mono.from(httpClient.exchange(request, EventSubscriptionVO.class))
                .doOnSuccess(res -> {
                    subscriptionHealthIndicator.setSubscriptionHealthy(entityType + eventType);
                    log.info("Successfully subscribed to {} {} events at {}", entityType, eventType, request.getUri());
                })
                .onErrorResume(t -> {
                    if (t instanceof HttpClientResponseException e) {
                        if (e.getStatus() == HttpStatus.CONFLICT) {
                            subscriptionHealthIndicator.setSubscriptionHealthy(entityType + eventType);
                            log.info("Subscription for {} {} already exists at {}", entityType, eventType, request.getUri());
                        } else {
                            String body = e.getResponse().getBody(String.class).orElse("<no body>");
                            log.warn("Event registration failed for {} at {} - Status: {} | Message: {} | Body: {} - retrying in {}s", entityType, request.getUri(), e.getStatus(), e.getMessage(), body, notificationProperties.getSubscriptionRetryInterval());
                            scheduleSubscription(notificationProperties.getSubscriptionRetryInterval(), entityType, eventType, apiAddress);
                        }
                        return Mono.empty();
                    }
                    log.warn("Could not create subscription for {} {} events at {} - retrying in {}s: {}", entityType, eventType,
                            request.getUri(), notificationProperties.getSubscriptionRetryInterval(), DownstreamError.reason(t));
                    scheduleSubscription(notificationProperties.getSubscriptionRetryInterval(), entityType, eventType, apiAddress);
                    return Mono.empty();
                }).subscribe();

    }
}

1	package org.fiware.iam.tmforum.notification;
2
3	import io.micronaut.context.annotation.Context;
4	import io.micronaut.context.annotation.Requires;
5	import io.micronaut.context.annotation.Value;
6	import io.micronaut.http.HttpMethod;
7	import io.micronaut.http.HttpRequest;
8	import io.micronaut.http.HttpStatus;
9	import io.micronaut.http.client.exceptions.HttpClientResponseException;
10	import io.micronaut.http.client.netty.DefaultHttpClient;
11	import io.micronaut.runtime.event.annotation.EventListener;
12	import io.micronaut.runtime.server.event.ServerStartupEvent;
13	import io.micronaut.scheduling.TaskScheduler;
14	import lombok.RequiredArgsConstructor;
15	import lombok.extern.slf4j.Slf4j;
16	import org.fiware.iam.logging.DownstreamError;
17	import org.fiware.iam.configuration.GeneralProperties;
18	import org.fiware.iam.configuration.NotificationProperties;
19	import org.fiware.iam.tmforum.party.model.EventSubscriptionInputVO;
20	import org.fiware.iam.tmforum.party.model.EventSubscriptionVO;
21	import reactor.core.publisher.Mono;
22
23	import java.time.Duration;
24	import java.util.List;
25	import java.util.Optional;
26
27	/**
28	* Responsible for subscribing to all configured tmforum events.
29	*/
30	@Requires(condition = NotificationProperties.NotificationCondition.class)
31	@Context
32	@RequiredArgsConstructor
33	@Slf4j	1 ✔
34	public class NotificationSubscriber {
35
36	private static final String QUERY_TEMPLATE = "eventType=%s%s";
37	private static final String LISTENER_ADDRESS_TEMPLATE = "%s/hub";
38	private static final String LISTENER_PATH = "/listener/event";
39
40	private final SubscriptionHealthIndicator subscriptionHealthIndicator;
41	private final NotificationProperties notificationProperties;
42	private final GeneralProperties generalProperties;
43	private final DefaultHttpClient httpClient;
44	private final TaskScheduler taskScheduler;
45
46	@Value("${micronaut.server.port:8080}")
47	private String servicePort;
48
49	private static String removeTrailingSlash(String path) {
50	if (path.endsWith("/")) {	1 ✔
51	return path.substring(0, path.length() - 1);	1 ✔
52	}
53	return path;	×
54	}
55
56	@EventListener
57	public void onApplicationEvent(ServerStartupEvent e) {
58	notificationProperties.getEntities()	1 ✔
59	.forEach(tmForumEntity ->	1 ✔
60	Optional.ofNullable(tmForumEntity.getEventTypes())	1 ✔
61	.orElse(List.of())	1 ✔
62	.forEach(eventType -> {	1 ✔
63	subscriptionHealthIndicator.initiateSubscriptionInMap(tmForumEntity.getEntityType() + eventType.getValue());	1 ✔
64	scheduleSubscription(notificationProperties.getSubscriptionInitialDelay(), tmForumEntity.getEntityType(), eventType.getValue(), tmForumEntity.getApiAddress());	1 ✔
65	}));	1 ✔
66	}	1 ✔
67
68	private void scheduleSubscription(long delaySeconds, String entityType, String eventType, String apiAddress) {
69	taskScheduler.schedule(Duration.ofSeconds(delaySeconds), () -> createSubscription(entityType, eventType, apiAddress));	1 ✔
70	}	1 ✔
71
72	public void createSubscription(String entityType, String eventType, String apiAddress) {
73	String callbackUrl = String.format("http://%s:%s%s%s", notificationProperties.getHost(), servicePort, removeTrailingSlash(generalProperties.getBasePath()), LISTENER_PATH);	1 ✔
74	log.debug("Attempting to register subscription for {} {} events at {}", entityType, eventType, String.format(LISTENER_ADDRESS_TEMPLATE, apiAddress));	1 ✔
75
76	EventSubscriptionInputVO subscription = new EventSubscriptionInputVO()	1 ✔
77	.callback(callbackUrl)	1 ✔
78	.query(String.format(QUERY_TEMPLATE, entityType, eventType));	1 ✔
79
80	HttpRequest<?> request = HttpRequest.create(HttpMethod.POST, String.format(LISTENER_ADDRESS_TEMPLATE, apiAddress)).body(subscription);	1 ✔
81
82	Mono.from(httpClient.exchange(request, EventSubscriptionVO.class))	1 ✔
83	.doOnSuccess(res -> {	1 ✔
84	subscriptionHealthIndicator.setSubscriptionHealthy(entityType + eventType);	×
85	log.info("Successfully subscribed to {} {} events at {}", entityType, eventType, request.getUri());	×
86	})	×
87	.onErrorResume(t -> {	1 ✔
88	if (t instanceof HttpClientResponseException e) {	×
89	if (e.getStatus() == HttpStatus.CONFLICT) {	×
90	subscriptionHealthIndicator.setSubscriptionHealthy(entityType + eventType);	×
91	log.info("Subscription for {} {} already exists at {}", entityType, eventType, request.getUri());	×
92	} else {
93	String body = e.getResponse().getBody(String.class).orElse("<no body>");	×
94	log.warn("Event registration failed for {} at {} - Status: {} \| Message: {} \| Body: {} - retrying in {}s", entityType, request.getUri(), e.getStatus(), e.getMessage(), body, notificationProperties.getSubscriptionRetryInterval());	×
95	scheduleSubscription(notificationProperties.getSubscriptionRetryInterval(), entityType, eventType, apiAddress);	×
96	}
97	return Mono.empty();	×
98	}
NEW 99	log.warn("Could not create subscription for {} {} events at {} - retrying in {}s: {}", entityType, eventType,	×
NEW 100	request.getUri(), notificationProperties.getSubscriptionRetryInterval(), DownstreamError.reason(t));	×
101	scheduleSubscription(notificationProperties.getSubscriptionRetryInterval(), entityType, eventType, apiAddress);	×
102	return Mono.empty();	×
103	}).subscribe();	1 ✔
104
105	}	1 ✔
106	}

FIWARE / contract-management / #104

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous