Skip to content

feat: configurable retry with backoff and service discovery #12

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
casc84ab merged 1 commit into from
Feb 13, 2026
Merged

feat: configurable retry with backoff and service discovery #12

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,13 @@
<description>A unified client library that provides reactive communication patterns for REST, SOAP, and gRPC services with circuit breaker support, health checks, and comprehensive monitoring capabilities.</description>

<dependencies>
<!-- Firefly Kernel (exception hierarchy, shared abstractions) -->
<dependency>
<groupId>org.fireflyframework</groupId>
<artifactId>fireflyframework-kernel</artifactId>
<version>${project.version}</version>
</dependency>

<!-- Spring Boot WebFlux -->
<dependency>
<groupId>org.springframework.boot</groupId>
Expand Down
44 changes: 43 additions & 1 deletion src/main/java/org/fireflyframework/client/builder/RestClientBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,13 @@ public class RestClientBuilder {
private WebClient webClient;
private CircuitBreakerManager circuitBreakerManager;

// Retry configuration
private boolean retryEnabled = true;
private int retryMaxAttempts = 3;
private Duration retryInitialBackoff = Duration.ofMillis(500);
private Duration retryMaxBackoff = Duration.ofSeconds(10);
private boolean retryJitterEnabled = true;

/**
* Creates a new REST client builder.
*
Expand Down Expand Up @@ -146,6 +153,36 @@ public RestClientBuilder circuitBreakerManager(CircuitBreakerManager circuitBrea
return this;
}

/**
* Configures retry behavior for the client.
*
* @param enabled whether retry is enabled
* @param maxAttempts maximum number of retry attempts
* @param initialBackoff initial wait duration before first retry
* @param maxBackoff maximum wait duration between retries
* @param jitterEnabled whether to add randomized jitter to backoff
* @return this builder
*/
public RestClientBuilder retry(boolean enabled, int maxAttempts, Duration initialBackoff,
Duration maxBackoff, boolean jitterEnabled) {
this.retryEnabled = enabled;
this.retryMaxAttempts = maxAttempts;
this.retryInitialBackoff = initialBackoff;
this.retryMaxBackoff = maxBackoff;
this.retryJitterEnabled = jitterEnabled;
return this;
}

/**
* Disables retry for the client.
*
* @return this builder
*/
public RestClientBuilder noRetry() {
this.retryEnabled = false;
return this;
}

/**
* Convenience method to set JSON content type headers.
*
Expand Down Expand Up @@ -179,7 +216,12 @@ public RestClient build() {
maxConnections,
defaultHeaders,
webClient,
circuitBreakerManager
circuitBreakerManager,
retryEnabled,
retryMaxAttempts,
retryInitialBackoff,
retryMaxBackoff,
retryJitterEnabled
);
}

Expand Down
155 changes: 140 additions & 15 deletions src/main/java/org/fireflyframework/client/discovery/ConsulServiceDiscoveryClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,14 +16,22 @@

package org.fireflyframework.client.discovery;

import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Collections;
import java.util.List;
import java.util.Map;

/**
* Consul-based service discovery client.
*
* <p>Integrates with HashiCorp Consul for service discovery.
*
* <p>Calls the Consul HTTP API to resolve healthy service instances, register, and deregister.
*
* @author Firefly Software Solutions Inc
* @since 1.0.0
Expand All @@ -32,49 +40,166 @@
public class ConsulServiceDiscoveryClient implements ServiceDiscoveryClient {

private final String consulUrl;
private final WebClient webClient;

public ConsulServiceDiscoveryClient(String consulUrl) {
this.consulUrl = consulUrl;
log.info("Initialized Consul Service Discovery with URL: {}", consulUrl);
this.consulUrl = consulUrl.endsWith("/") ? consulUrl.substring(0, consulUrl.length() - 1) : consulUrl;
this.webClient = WebClient.builder()
.baseUrl(this.consulUrl)
.defaultHeader("Accept", "application/json")
.defaultHeader("Content-Type", "application/json")
.build();
log.info("Initialized Consul Service Discovery with URL: {}", this.consulUrl);
}

@Override
public Mono<String> resolveEndpoint(String serviceName) {
// TODO: Implement actual Consul API call
log.debug("Resolving endpoint for service: {} via Consul", serviceName);
return Mono.just("http://" + serviceName.toLowerCase());
return getHealthyInstance(serviceName)
.map(ServiceInstance::getUri)
.switchIfEmpty(Mono.defer(() -> {
log.warn("No healthy instance found for service '{}' in Consul, falling back to name-based resolution",
serviceName);
return Mono.just("http://" + serviceName.toLowerCase());
}));
}

@Override
public Flux<ServiceInstance> getInstances(String serviceName) {
// TODO: Implement actual Consul API call
return Flux.empty();
log.debug("Getting instances for service: {} from Consul at {}", serviceName, consulUrl);
return webClient.get()
.uri("/v1/health/service/{serviceName}?passing=true", serviceName)
.retrieve()
.bodyToFlux(ConsulHealthServiceEntry.class)
.map(entry -> {
ConsulService svc = entry.getService();
ConsulCheck[] checks = entry.getChecks();

HealthStatus status = HealthStatus.UP;
if (checks != null) {
for (ConsulCheck check : checks) {
if (!"passing".equalsIgnoreCase(check.getStatus())) {
status = HealthStatus.DOWN;
break;
}
}
}

return new ServiceInstance(
svc.getId(),
serviceName,
svc.getAddress() != null ? svc.getAddress() : entry.getNode().getAddress(),
svc.getPort(),
false,
status,
svc.getMeta() != null ? svc.getMeta() : Map.of()
);
})
.onErrorResume(e -> {
log.error("Failed to get instances for service '{}' from Consul: {}", serviceName, e.getMessage());
return Flux.empty();
});
}

@Override
public Mono<ServiceInstance> getHealthyInstance(String serviceName) {
return getInstances(serviceName)
.filter(ServiceInstance::isHealthy)
.next();
.filter(ServiceInstance::isHealthy)
.next();
}

@Override
public Mono<Void> register(ServiceInstance instance) {
// TODO: Implement Consul registration
log.info("Registering instance {} with Consul", instance.instanceId());
return Mono.empty();
log.info("Registering instance {} with Consul for service {}", instance.instanceId(), instance.serviceName());

ConsulServiceRegistration reg = new ConsulServiceRegistration();
reg.setId(instance.instanceId());
reg.setName(instance.serviceName());
reg.setAddress(instance.host());
reg.setPort(instance.port());
reg.setMeta(instance.metadata());

return webClient.put()
.uri("/v1/agent/service/register")
.bodyValue(reg)
.retrieve()
.toBodilessEntity()
.doOnSuccess(r -> log.info("Successfully registered {} with Consul", instance.instanceId()))
.doOnError(e -> log.error("Failed to register {} with Consul: {}", instance.instanceId(), e.getMessage()))
.then();
}

@Override
public Mono<Void> deregister(String instanceId) {
// TODO: Implement Consul deregistration
log.info("Deregistering instance {} from Consul", instanceId);
return Mono.empty();
return webClient.put()
.uri("/v1/agent/service/deregister/{serviceId}", instanceId)
.retrieve()
.toBodilessEntity()
.doOnSuccess(r -> log.info("Successfully deregistered {} from Consul", instanceId))
.doOnError(e -> log.error("Failed to deregister {} from Consul: {}", instanceId, e.getMessage()))
.then();
}

@Override
public Mono<Boolean> isServiceAvailable(String serviceName) {
return getInstances(serviceName).hasElements();
}

// --- Consul REST API DTOs ---

@Data
@JsonIgnoreProperties(ignoreUnknown = true)
static class ConsulHealthServiceEntry {
@JsonProperty("Node")
private ConsulNode node;
@JsonProperty("Service")
private ConsulService service;
@JsonProperty("Checks")
private ConsulCheck[] checks;
}

@Data
@JsonIgnoreProperties(ignoreUnknown = true)
static class ConsulNode {
@JsonProperty("Address")
private String address;
}

@Data
@JsonIgnoreProperties(ignoreUnknown = true)
static class ConsulService {
@JsonProperty("ID")
private String id;
@JsonProperty("Service")
private String service;
@JsonProperty("Address")
private String address;
@JsonProperty("Port")
private int port;
@JsonProperty("Meta")
private Map<String, String> meta;
}

@Data
@JsonIgnoreProperties(ignoreUnknown = true)
static class ConsulCheck {
@JsonProperty("Status")
private String status;
}

@Data
static class ConsulServiceRegistration {
@JsonProperty("ID")
private String id;
@JsonProperty("Name")
private String name;
@JsonProperty("Address")
private String address;
@JsonProperty("Port")
private int port;
@JsonProperty("Meta")
private Map<String, String> meta;
}
}

Loading