Replication API Quick Reference

try {
    replicator.replicate(session, ReplicationActionType.ACTIVATE, path);
} catch (ReplicationException e) {
    LOG.error("Replication failed for path: {}", path, e);
    // Handle error (retry, alert, log)
}

@Reference
private Replicator replicator;

public void activate(ResourceResolver resolver, String path) 
    throws ReplicationException {
    Session session = resolver.adaptTo(Session.class);
    if (session == null) {
        throw new IllegalStateException("Unable to adapt to Session");
    }
    replicator.replicate(session, ReplicationActionType.ACTIVATE, path);
}

public void activateSync(ResourceResolver resolver, String path) 
    throws ReplicationException {
    Session session = resolver.adaptTo(Session.class);
    if (session == null) {
        throw new IllegalStateException("Unable to adapt to Session");
    }
    
    ReplicationOptions opts = new ReplicationOptions();
    opts.setSynchronous(true);
    
    replicator.replicate(session, ReplicationActionType.ACTIVATE, path, opts);
}

public void activateBulk(ResourceResolver resolver, List<String> paths) 
    throws ReplicationException {
    Session session = resolver.adaptTo(Session.class);
    if (session == null) {
        throw new IllegalStateException("Unable to adapt to Session");
    }
    
    ReplicationOptions opts = new ReplicationOptions();
    opts.setSynchronous(false); // Async for performance
    
    String[] pathArray = paths.toArray(new String[0]);
    replicator.replicate(session, ReplicationActionType.ACTIVATE, pathArray, opts);
}

public void activateToSpecificAgent(ResourceResolver resolver, String path, 
                                     String targetAgentId) 
    throws ReplicationException {
    Session session = resolver.adaptTo(Session.class);
    if (session == null) {
        throw new IllegalStateException("Unable to adapt to Session");
    }
    
    ReplicationOptions opts = new ReplicationOptions();
    opts.setFilter(agent -> agent.getId().equals(targetAgentId));
    
    replicator.replicate(session, ReplicationActionType.ACTIVATE, path, opts);
}

public boolean isActivated(ResourceResolver resolver, String path) {
    Session session = resolver.adaptTo(Session.class);
    if (session == null) {
        return false;
    }
    
    ReplicationStatus status = replicator.getReplicationStatus(session, path);
    return status != null && status.isActivated();
}

public boolean canActivate(ResourceResolver resolver, String path) {
    Session session = resolver.adaptTo(Session.class);
    if (session == null) {
        return false;
    }
    
    try {
        replicator.checkPermission(session, ReplicationActionType.ACTIVATE, path);
        return true;
    } catch (ReplicationException e) {
        return false; // User lacks permission
    }
}

@Reference
private ResourceResolverFactory resolverFactory;

@Reference
private Replicator replicator;

public void activateWithServiceUser(String path) throws ReplicationException {
    Map<String, Object> authInfo = new HashMap<>();
    authInfo.put(ResourceResolverFactory.SUBSERVICE, "replication-service");
    
    try (ResourceResolver resolver = resolverFactory.getServiceResourceResolver(authInfo)) {
        Session session = resolver.adaptTo(Session.class);
        if (session == null) {
            throw new IllegalStateException("Unable to adapt to Session");
        }
        replicator.replicate(session, ReplicationActionType.ACTIVATE, path);
    } catch (LoginException e) {
        LOG.error("Service user login failed", e);
        throw new ReplicationException("Authentication failed", e);
    }
}

public void activateWithRetry(ResourceResolver resolver, String path, int maxRetries) 
    throws ReplicationException {
    
    Session session = resolver.adaptTo(Session.class);
    if (session == null) {
        throw new IllegalStateException("Unable to adapt to Session");
    }
    
    int attempt = 0;
    long delayMs = 1000; // Start with 1 second
    
    while (attempt < maxRetries) {
        try {
            replicator.replicate(session, ReplicationActionType.ACTIVATE, path);
            LOG.info("Replication succeeded on attempt {}", attempt + 1);
            return; // Success
            
        } catch (ReplicationException e) {
            attempt++;
            
            if (attempt >= maxRetries) {
                LOG.error("Replication failed after {} attempts", maxRetries, e);
                throw e; // Give up after max retries
            }
            
            // Exponential backoff: 1s, 2s, 4s, 8s, etc.
            LOG.warn("Replication attempt {} failed, retrying in {}ms", attempt, delayMs, e);
            
            try {
                Thread.sleep(delayMs);
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
                throw new ReplicationException("Retry interrupted", ie);
            }
            
            delayMs *= 2; // Double the delay for next attempt
        }
    }
}

public void activateWithFallback(ResourceResolver resolver, String path) 
    throws ReplicationException {
    
    Session session = resolver.adaptTo(Session.class);
    if (session == null) {
        throw new IllegalStateException("Unable to adapt to Session");
    }
    
    // Try synchronous first
    ReplicationOptions syncOpts = new ReplicationOptions();
    syncOpts.setSynchronous(true);
    
    try {
        replicator.replicate(session, ReplicationActionType.ACTIVATE, path, syncOpts);
        LOG.info("Synchronous replication completed");
        
    } catch (ReplicationException e) {
        // Check if it's a timeout or transient error
        if (isTransientError(e)) {
            LOG.warn("Synchronous replication failed, falling back to async", e);
            
            // Fallback to asynchronous
            ReplicationOptions asyncOpts = new ReplicationOptions();
            asyncOpts.setSynchronous(false);
            
            replicator.replicate(session, ReplicationActionType.ACTIVATE, path, asyncOpts);
            LOG.info("Asynchronous replication queued as fallback");
        } else {
            // Permanent error, don't retry
            throw e;
        }
    }
}

private boolean isTransientError(ReplicationException e) {
    String message = e.getMessage();
    return message != null && (
        message.contains("timeout") ||
        message.contains("Connection refused") ||
        message.contains("SocketTimeoutException")
    );
}

public class ReplicationCircuitBreaker {
    
    private static final int FAILURE_THRESHOLD = 5;
    private static final long TIMEOUT_MS = 60000; // 1 minute
    
    private int failureCount = 0;
    private long lastFailureTime = 0;
    private boolean circuitOpen = false;
    
    @Reference
    private Replicator replicator;
    
    public void activateWithCircuitBreaker(ResourceResolver resolver, String path) 
        throws ReplicationException {
        
        // Check if circuit is open
        if (circuitOpen) {
            if (System.currentTimeMillis() - lastFailureTime < TIMEOUT_MS) {
                throw new ReplicationException("Circuit breaker open - replication disabled temporarily");
            } else {
                // Timeout expired, try to close circuit
                circuitOpen = false;
                failureCount = 0;
                LOG.info("Circuit breaker reset, attempting replication");
            }
        }
        
        Session session = resolver.adaptTo(Session.class);
        if (session == null) {
            throw new IllegalStateException("Unable to adapt to Session");
        }
        
        try {
            replicator.replicate(session, ReplicationActionType.ACTIVATE, path);
            
            // Success - reset failure count
            failureCount = 0;
            LOG.info("Replication successful");
            
        } catch (ReplicationException e) {
            failureCount++;
            lastFailureTime = System.currentTimeMillis();
            
            if (failureCount >= FAILURE_THRESHOLD) {
                circuitOpen = true;
                LOG.error("Circuit breaker opened after {} failures", failureCount);
            }
            
            throw e;
        }
    }
}