Service Discovery and Registry

/**
 * Registry interface for service discovery operations
 */
@SPI("dubbo")
public interface Registry extends Node, RegistryService {
    // Inherits methods from RegistryService and Node interfaces
}

/**
 * Registry service operations interface
 */
public interface RegistryService {
    /**
     * Register service provider
     * @param url Service provider URL
     */
    void register(URL url);
    
    /**
     * Unregister service provider
     * @param url Service provider URL
     */
    void unregister(URL url);
    
    /**
     * Subscribe to service changes
     * @param url Consumer URL with subscription parameters
     * @param listener Notification listener for service changes
     */
    void subscribe(URL url, NotifyListener listener);
    
    /**
     * Unsubscribe from service changes
     * @param url Consumer URL
     * @param listener Notification listener to remove
     */
    void unsubscribe(URL url, NotifyListener listener);
    
    /**
     * Lookup available providers
     * @param url Service URL to lookup
     * @return List of available provider URLs
     */
    List<URL> lookup(URL url);
}

/**
 * Node interface for lifecycle management
 */
public interface Node {
    /** Get node URL */
    URL getUrl();
    
    /** Check if node is available */
    boolean isAvailable();
    
    /** Destroy node and release resources */
    void destroy();
}

/**
 * Registry factory for creating registry instances
 */
@SPI("dubbo")
public interface RegistryFactory {
    /**
     * Create registry instance
     * @param url Registry URL with connection parameters
     * @return Registry instance
     */
    Registry getRegistry(URL url);
}

/**
 * Abstract registry factory with common functionality
 */
public abstract class AbstractRegistryFactory implements RegistryFactory {
    /** Registry cache by URL */
    private static final Map<String, Registry> REGISTRIES = new ConcurrentHashMap<>();
    
    @Override
    public Registry getRegistry(URL url) {
        // Normalize URL and get from cache or create new
        url = URLBuilder.from(url)
            .setPath(RegistryService.class.getName())
            .addParameter(INTERFACE_KEY, RegistryService.class.getName())
            .removeParameters(EXPORT_KEY, REFER_KEY)
            .build();
            
        String key = url.toServiceStringWithoutResolving();
        Registry registry = REGISTRIES.get(key);
        if (registry != null) {
            return registry;
        }
        
        synchronized (this) {
            registry = REGISTRIES.get(key);
            if (registry == null) {
                registry = createRegistry(url);
                REGISTRIES.put(key, registry);
            }
        }
        return registry;
    }
    
    /** Template method for creating specific registry implementation */
    protected abstract Registry createRegistry(URL url);
}

/**
 * Notification listener for registry changes
 */
public interface NotifyListener {
    /**
     * Handle provider list changes
     * @param urls Updated list of provider URLs
     */
    void notify(List<URL> urls);
}

/**
 * Registry event for service changes
 */
public class RegistryNotification {
    public enum EventType {
        REGISTER,
        UNREGISTER,
        UPDATE
    }
    
    private final EventType eventType;
    private final URL url;
    private final long timestamp;
    
    public RegistryNotification(EventType eventType, URL url) {
        this.eventType = eventType;
        this.url = url;
        this.timestamp = System.currentTimeMillis();
    }
    
    public EventType getEventType() { return eventType; }
    public URL getUrl() { return url; }
    public long getTimestamp() { return timestamp; }
}

/**
 * ZooKeeper registry implementation
 */
public class ZookeeperRegistry extends FailbackRegistry {
    public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter);
    
    /** Get ZooKeeper client */
    public ZookeeperClient getZookeeperClient();
    
    /** Create ZooKeeper path */
    protected void createPersistent(String path);
    
    /** Create ephemeral ZooKeeper path */
    protected void createEphemeral(String path);
    
    /** Add child listener */
    protected void addChildListener(String path, ChildListener listener);
    
    /** Remove child listener */
    protected void removeChildListener(String path, ChildListener listener);
    
    @Override
    protected void doRegister(URL url) {
        // Create ephemeral sequential node for provider
        String path = toUrlPath(url);
        createEphemeral(path);
    }
    
    @Override
    protected void doUnregister(URL url) {
        // Delete ZooKeeper node
        String path = toUrlPath(url);
        zkClient.delete(path);
    }
    
    @Override
    protected void doSubscribe(URL url, NotifyListener listener) {
        // Subscribe to ZooKeeper path changes
        String path = toCategoryPath(url);
        addChildListener(path, new ZookeeperChildListener(listener));
    }
}

/**
 * ZooKeeper registry factory
 */
public class ZookeeperRegistryFactory extends AbstractRegistryFactory {
    private ZookeeperTransporter zookeeperTransporter;
    
    @Override
    protected Registry createRegistry(URL url) {
        return new ZookeeperRegistry(url, zookeeperTransporter);
    }
}

// ZooKeeper registry configuration
RegistryConfig registry = new RegistryConfig();
registry.setAddress("zookeeper://127.0.0.1:2181");
registry.setUsername("admin");
registry.setPassword("password");
registry.setTimeout(5000);
registry.setSession(60000);

// Cluster ZooKeeper setup
registry.setAddress("zookeeper://127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183");

// With namespace
registry.setAddress("zookeeper://127.0.0.1:2181/dubbo");

/**
 * Nacos registry implementation
 */
public class NacosRegistry extends FailbackRegistry {
    public NacosRegistry(URL url, NacosNamingService nacosNamingService);
    
    /** Get Nacos naming service */
    public NacosNamingService getNacosNamingService();
    
    @Override
    protected void doRegister(URL url) {
        // Register service instance to Nacos
        Instance instance = createNacosInstance(url);
        nacosNamingService.registerInstance(getServiceName(url), instance);
    }
    
    @Override
    protected void doUnregister(URL url) {
        // Deregister service instance from Nacos
        Instance instance = createNacosInstance(url);
        nacosNamingService.deregisterInstance(getServiceName(url), instance);
    }
    
    @Override
    protected void doSubscribe(URL url, NotifyListener listener) {
        // Subscribe to Nacos service changes
        String serviceName = getServiceName(url);
        nacosNamingService.subscribe(serviceName, new NacosEventListener(listener));
    }
    
    /** Convert Dubbo URL to Nacos instance */
    private Instance createNacosInstance(URL url) {
        Instance instance = new Instance();
        instance.setIp(url.getHost());
        instance.setPort(url.getPort());
        instance.setWeight(url.getParameter(WEIGHT_KEY, DEFAULT_WEIGHT));
        instance.setHealthy(true);
        instance.setEnabled(true);
        instance.setMetadata(new HashMap<>(url.getParameters()));
        return instance;
    }
}

/**
 * Nacos registry factory
 */
public class NacosRegistryFactory extends AbstractRegistryFactory {
    @Override
    protected Registry createRegistry(URL url) {
        return new NacosRegistry(url, buildNacosNamingService(url));
    }
}

// Nacos registry configuration
RegistryConfig registry = new RegistryConfig();
registry.setAddress("nacos://127.0.0.1:8848");
registry.setGroup("DEFAULT_GROUP");
registry.setNamespace("public");

// With configuration parameters
registry.setAddress("nacos://127.0.0.1:8848?namespace=dev&group=dubbo");

// Authentication
registry.setUsername("nacos");
registry.setPassword("nacos");

/**
 * Consul registry implementation
 */
public class ConsulRegistry extends FailbackRegistry {
    public ConsulRegistry(URL url, ConsulClient consulClient);
    
    /** Get Consul client */
    public ConsulClient getConsulClient();
    
    @Override
    protected void doRegister(URL url) {
        // Register service to Consul
        NewService service = createConsulService(url);
        consulClient.agentServiceRegister(service);
    }
    
    @Override
    protected void doUnregister(URL url) {
        // Deregister service from Consul
        String serviceId = buildServiceId(url);
        consulClient.agentServiceDeregister(serviceId);
    }
    
    @Override
    protected void doSubscribe(URL url, NotifyListener listener) {
        // Watch Consul service changes
        String serviceName = getServiceName(url);
        consulClient.healthServicesPass(serviceName, new ConsulQueryCallback(listener));
    }
    
    /** Convert Dubbo URL to Consul service */
    private NewService createConsulService(URL url) {
        NewService service = new NewService();
        service.setId(buildServiceId(url));
        service.setName(getServiceName(url));
        service.setAddress(url.getHost());
        service.setPort(url.getPort());
        service.setTags(buildTags(url));
        service.setCheck(createHealthCheck(url));
        return service;
    }
}

/**
 * Etcd registry implementation
 */
public class EtcdRegistry extends FailbackRegistry {
    public EtcdRegistry(URL url, EtcdClient etcdClient);
    
    @Override
    protected void doRegister(URL url) {
        // Register service to etcd with TTL
        String key = buildEtcdKey(url);
        String value = url.toFullString();
        etcdClient.putEphemeral(key, value);
    }
    
    @Override
    protected void doUnregister(URL url) {
        // Remove service from etcd
        String key = buildEtcdKey(url);
        etcdClient.delete(key);
    }
    
    @Override
    protected void doSubscribe(URL url, NotifyListener listener) {
        // Watch etcd key prefix changes
        String keyPrefix = buildEtcdKeyPrefix(url);
        etcdClient.watchPrefix(keyPrefix, new EtcdWatchListener(listener));
    }
}

/**
 * Multicast registry for local network discovery
 */
public class MulticastRegistry extends FailbackRegistry {
    public MulticastRegistry(URL url);
    
    /** Get multicast socket */
    public MulticastSocket getMulticastSocket();
    
    @Override
    protected void doRegister(URL url) {
        // Broadcast registration message
        String message = REGISTER + url.toFullString();
        broadcast(message);
    }
    
    @Override
    protected void doUnregister(URL url) {
        // Broadcast unregistration message
        String message = UNREGISTER + url.toFullString();
        broadcast(message);
    }
    
    @Override
    protected void doSubscribe(URL url, NotifyListener listener) {
        // Add subscription listener
        subscriptions.put(url, listener);
        
        // Request current providers
        String message = SUBSCRIBE + url.toFullString();
        broadcast(message);
    }
    
    /** Broadcast message to multicast group */
    private void broadcast(String message) {
        byte[] data = message.getBytes();
        DatagramPacket packet = new DatagramPacket(data, data.length, multicastAddress, multicastPort);
        multicastSocket.send(packet);
    }
}

// Multicast registry (development/testing)
RegistryConfig registry = new RegistryConfig();
registry.setAddress("multicast://224.5.6.7:1234");

// Custom multicast parameters
registry.setAddress("multicast://239.255.255.255:1234?timeout=3000&cleanup=60000");

/**
 * Registry configuration constants
 */
public class RegistryConstants {
    /** Registry protocol */
    public static final String REGISTRY_KEY = "registry";
    public static final String REGISTRY_PROTOCOL = "registry";
    
    /** Dynamic registration */
    public static final String DYNAMIC_KEY = "dynamic";
    public static final boolean DEFAULT_DYNAMIC = true;
    
    /** Service categories */
    public static final String CATEGORY_KEY = "category";
    public static final String PROVIDERS_CATEGORY = "providers";
    public static final String CONSUMERS_CATEGORY = "consumers";
    public static final String ROUTERS_CATEGORY = "routers";
    public static final String CONFIGURATORS_CATEGORY = "configurators";
    
    /** Registry check */
    public static final String REGISTRY_CHECK_KEY = "check";
    
    /** Registry retry period */
    public static final String REGISTRY_RETRY_PERIOD_KEY = "retry.period";
    public static final int DEFAULT_REGISTRY_RETRY_PERIOD = 5000;
    
    /** Registry session timeout */
    public static final String SESSION_TIMEOUT_KEY = "session";
    public static final int DEFAULT_SESSION_TIMEOUT = 60000;
}

/**
 * Failback registry providing automatic retry for failed operations
 */
public abstract class FailbackRegistry extends AbstractRegistry {
    /** Failed registration URLs */
    private final Set<URL> failedRegistered = new ConcurrentHashSet<>();
    
    /** Failed unregistration URLs */
    private final Set<URL> failedUnregistered = new ConcurrentHashSet<>();
    
    /** Failed subscriptions */
    private final Map<URL, Set<NotifyListener>> failedSubscribed = new ConcurrentHashMap<>();
    
    /** Failed unsubscriptions */  
    private final Map<URL, Set<NotifyListener>> failedUnsubscribed = new ConcurrentHashMap<>();
    
    public FailbackRegistry(URL url) {
        super(url);
        // Start retry timer
        this.retryTimer = new HashedWheelTimer(new NamedThreadFactory("DubboRegistryFailedRetryTimer", true), 
                                              retryPeriod, TimeUnit.MILLISECONDS, 128);
        this.retryTimer.newTimeout(this::retry, retryPeriod, TimeUnit.MILLISECONDS);
    }
    
    @Override
    public void register(URL url) {
        super.register(url);
        failedRegistered.remove(url);
        failedUnregistered.remove(url);
        try {
            doRegister(url);
        } catch (Exception e) {
            failedRegistered.add(url);
            throw new RpcException("Failed to register " + url, e);
        }
    }
    
    @Override
    public void unregister(URL url) {
        super.unregister(url);
        failedRegistered.remove(url);
        failedUnregistered.remove(url);
        try {
            doUnregister(url);
        } catch (Exception e) {
            failedUnregistered.add(url);
            throw new RpcException("Failed to unregister " + url, e);
        }
    }
    
    /** Retry failed operations */
    private void retry(Timeout timeout) {
        // Retry failed registrations
        if (!failedRegistered.isEmpty()) {
            Set<URL> failed = new HashSet<>(failedRegistered);
            for (URL url : failed) {
                try {
                    doRegister(url);
                    failedRegistered.remove(url);
                } catch (Exception e) {
                    // Keep in failed set for next retry
                }
            }
        }
        
        // Schedule next retry
        retryTimer.newTimeout(this::retry, retryPeriod, TimeUnit.MILLISECONDS);
    }
    
    /** Template method for actual registration */
    protected abstract void doRegister(URL url);
    
    /** Template method for actual unregistration */
    protected abstract void doUnregister(URL url);
    
    /** Template method for actual subscription */
    protected abstract void doSubscribe(URL url, NotifyListener listener);
    
    /** Template method for actual unsubscription */
    protected abstract void doUnsubscribe(URL url, NotifyListener listener);
}

/**
 * Service instance representation
 */
public interface ServiceInstance extends Serializable {
    /** Get service name */
    String getServiceName();
    
    /** Get instance host */
    String getHost();
    
    /** Get instance port */
    int getPort();
    
    /** Check if instance is enabled */
    boolean isEnabled();
    
    /** Check if instance is healthy */
    boolean isHealthy();
    
    /** Get instance metadata */
    Map<String, String> getMetadata();
    
    /** Get instance ID */
    String getId();
}

/**
 * Service discovery interface for cloud-native service mesh
 */
@SPI("zookeeper")
public interface ServiceDiscovery extends Closeable {
    /**
     * Initialize service discovery
     * @throws Exception if initialization fails
     */
    void initialize(URL registryURL) throws Exception;
    
    /**
     * Register service instance
     * @param serviceInstance Service instance to register
     * @throws RuntimeException if registration fails
     */
    void register(ServiceInstance serviceInstance) throws RuntimeException;
    
    /**
     * Update service instance
     * @param serviceInstance Service instance to update
     * @throws RuntimeException if update fails
     */
    void update(ServiceInstance serviceInstance) throws RuntimeException;
    
    /**
     * Unregister service instance
     * @param serviceInstance Service instance to unregister
     * @throws RuntimeException if unregistration fails
     */
    void unregister(ServiceInstance serviceInstance) throws RuntimeException;
    
    /**
     * Get service instances by service name
     * @param serviceName Service name
     * @return Set of service instances
     * @throws RuntimeException if lookup fails
     */
    Set<ServiceInstance> getInstances(String serviceName) throws RuntimeException;
    
    /**
     * Get all service names
     * @return Set of service names
     * @throws RuntimeException if lookup fails
     */
    Set<String> getServices() throws RuntimeException;
    
    /**
     * Add service instance change listener
     * @param serviceName Service name to watch
     * @param listener Change listener
     * @throws RuntimeException if listener addition fails
     */
    void addServiceInstancesChangedListener(String serviceName, 
                                           ServiceInstancesChangedListener listener) 
                                           throws RuntimeException;
}

// Complete registry setup with multiple backends
DubboBootstrap bootstrap = DubboBootstrap.getInstance();

// Primary registry (ZooKeeper)
RegistryConfig primaryRegistry = new RegistryConfig();
primaryRegistry.setAddress("zookeeper://127.0.0.1:2181");
primaryRegistry.setGroup("production");

// Backup registry (Nacos)
RegistryConfig backupRegistry = new RegistryConfig();
backupRegistry.setAddress("nacos://127.0.0.1:8848");
backupRegistry.setGroup("production");
backupRegistry.setBackup(true);

bootstrap.registries(Arrays.asList(primaryRegistry, backupRegistry));

// Service with multiple registry support
ServiceConfig<GreeterService> service = new ServiceConfig<>();
service.setInterface(GreeterService.class);
service.setRef(new GreeterServiceImpl());
service.setRegistry(primaryRegistry);
service.export();