dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/support/AbstractRegistry.java - dubbo-hessian-lite - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package com.alibaba.dubbo.registry.support;

 import com.alibaba.dubbo.common.Constants;
 import com.alibaba.dubbo.common.URL;
 import com.alibaba.dubbo.common.logger.Logger;
 import com.alibaba.dubbo.common.logger.LoggerFactory;
 import com.alibaba.dubbo.common.utils.ConcurrentHashSet;
 import com.alibaba.dubbo.common.utils.ConfigUtils;
 import com.alibaba.dubbo.common.utils.NamedThreadFactory;
 import com.alibaba.dubbo.common.utils.UrlUtils;
 import com.alibaba.dubbo.registry.NotifyListener;
 import com.alibaba.dubbo.registry.Registry;

 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.RandomAccessFile;
 import java.nio.channels.FileChannel;
 import java.nio.channels.FileLock;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;

 /**
  * AbstractRegistry. (SPI, Prototype, ThreadSafe)
  *
  */
 public abstract class AbstractRegistry implements Registry {

     // URL address separator, used in file cache, service provider URL separation
     private static final char URL_SEPARATOR = ' ';
     // URL address separated regular expression for parsing the service provider URL list in the file cache
     private static final String URL_SPLIT = "\\s+";
     // Log output
     protected final Logger logger = LoggerFactory.getLogger(getClass());
     // Local disk cache, where the special key value.registies records the list of registry centers, and the others are the list of notified service providers
     private final Properties properties = new Properties();
     // File cache timing writing
     private final ExecutorService registryCacheExecutor = Executors.newFixedThreadPool(1, new NamedThreadFactory("DubboSaveRegistryCache", true));
     // Is it synchronized to save the file
     private final boolean syncSaveFile;
     private final AtomicLong lastCacheChanged = new AtomicLong();
     private final Set<URL> registered = new ConcurrentHashSet<URL>();
     private final ConcurrentMap<URL, Set<NotifyListener>> subscribed = new ConcurrentHashMap<URL, Set<NotifyListener>>();
     private final ConcurrentMap<URL, Map<String, List<URL>>> notified = new ConcurrentHashMap<URL, Map<String, List<URL>>>();
     private URL registryUrl;
     // Local disk cache file
     private File file;

     public AbstractRegistry(URL url) {
         setUrl(url);
         // Start file save timer
         syncSaveFile = url.getParameter(Constants.REGISTRY_FILESAVE_SYNC_KEY, false);
         String filename = url.getParameter(Constants.FILE_KEY, System.getProperty("user.home") + "/.dubbo/dubbo-registry-" + url.getParameter(Constants.APPLICATION_KEY) + "-" + url.getAddress() + ".cache");
         File file = null;
         if (ConfigUtils.isNotEmpty(filename)) {
             file = new File(filename);
             if (!file.exists() && file.getParentFile() != null && !file.getParentFile().exists()) {
                 if (!file.getParentFile().mkdirs()) {
                     throw new IllegalArgumentException("Invalid registry store file " + file + ", cause: Failed to create directory " + file.getParentFile() + "!");
                 }
             }
         }
         this.file = file;
         loadProperties();
         notify(url.getBackupUrls());
     }

     protected static List<URL> filterEmpty(URL url, List<URL> urls) {
         if (urls == null || urls.isEmpty()) {
             List<URL> result = new ArrayList<URL>(1);
             result.add(url.setProtocol(Constants.EMPTY_PROTOCOL));
             return result;
         }
         return urls;
     }

     @Override
     public URL getUrl() {
         return registryUrl;
     }

     protected void setUrl(URL url) {
         if (url == null) {
             throw new IllegalArgumentException("registry url == null");
         }
         this.registryUrl = url;
     }

     public Set<URL> getRegistered() {
         return registered;
     }

     public Map<URL, Set<NotifyListener>> getSubscribed() {
         return subscribed;
     }

     public Map<URL, Map<String, List<URL>>> getNotified() {
         return notified;
     }

     public File getCacheFile() {
         return file;
     }

     public Properties getCacheProperties() {
         return properties;
     }

     public AtomicLong getLastCacheChanged() {
         return lastCacheChanged;
     }

     public void doSaveProperties(long version) {
         if (version < lastCacheChanged.get()) {
             return;
         }
         if (file == null) {
             return;
         }
         // Save
         try {
             File lockfile = new File(file.getAbsolutePath() + ".lock");
             if (!lockfile.exists()) {
                 lockfile.createNewFile();
             }
             RandomAccessFile raf = new RandomAccessFile(lockfile, "rw");
             try {
                 FileChannel channel = raf.getChannel();
                 try {
                     FileLock lock = channel.tryLock();
                     if (lock == null) {
                         throw new IOException("Can not lock the registry cache file " + file.getAbsolutePath() + ", ignore and retry later, maybe multi java process use the file, please config: dubbo.registry.file=xxx.properties");
                     }
                     // Save
                     try {
                         if (!file.exists()) {
                             file.createNewFile();
                         }
                         FileOutputStream outputFile = new FileOutputStream(file);
                         try {
                             properties.store(outputFile, "Dubbo Registry Cache");
                         } finally {
                             outputFile.close();
                         }
                     } finally {
                         lock.release();
                     }
                 } finally {
                     channel.close();
                 }
             } finally {
                 raf.close();
             }
         } catch (Throwable e) {
             if (version < lastCacheChanged.get()) {
                 return;
             } else {
                 registryCacheExecutor.execute(new SaveProperties(lastCacheChanged.incrementAndGet()));
             }
             logger.warn("Failed to save registry store file, cause: " + e.getMessage(), e);
         }
     }

     private void loadProperties() {
         if (file != null && file.exists()) {
             InputStream in = null;
             try {
                 in = new FileInputStream(file);
                 properties.load(in);
                 if (logger.isInfoEnabled()) {
                     logger.info("Load registry store file " + file + ", data: " + properties);
                 }
             } catch (Throwable e) {
                 logger.warn("Failed to load registry store file " + file, e);
             } finally {
                 if (in != null) {
                     try {
                         in.close();
                     } catch (IOException e) {
                         logger.warn(e.getMessage(), e);
                     }
                 }
             }
         }
     }

     public List<URL> getCacheUrls(URL url) {
         for (Map.Entry<Object, Object> entry : properties.entrySet()) {
             String key = (String) entry.getKey();
             String value = (String) entry.getValue();
             if (key != null && key.length() > 0 && key.equals(url.getServiceKey())
                     && (Character.isLetter(key.charAt(0)) || key.charAt(0) == '_')
                     && value != null && value.length() > 0) {
                 String[] arr = value.trim().split(URL_SPLIT);
                 List<URL> urls = new ArrayList<URL>();
                 for (String u : arr) {
                     urls.add(URL.valueOf(u));
                 }
                 return urls;
             }
         }
         return null;
     }

     @Override
     public List<URL> lookup(URL url) {
         List<URL> result = new ArrayList<URL>();
         Map<String, List<URL>> notifiedUrls = getNotified().get(url);
         if (notifiedUrls != null && notifiedUrls.size() > 0) {
             for (List<URL> urls : notifiedUrls.values()) {
                 for (URL u : urls) {
                     if (!Constants.EMPTY_PROTOCOL.equals(u.getProtocol())) {
                         result.add(u);
                     }
                 }
             }
         } else {
             final AtomicReference<List<URL>> reference = new AtomicReference<List<URL>>();
             NotifyListener listener = new NotifyListener() {
                 @Override
                 public void notify(List<URL> urls) {
                     reference.set(urls);
                 }
             };
             subscribe(url, listener); // Subscribe logic guarantees the first notify to return
             List<URL> urls = reference.get();
             if (urls != null && !urls.isEmpty()) {
                 for (URL u : urls) {
                     if (!Constants.EMPTY_PROTOCOL.equals(u.getProtocol())) {
                         result.add(u);
                     }
                 }
             }
         }
         return result;
     }

     @Override
     public void register(URL url) {
         if (url == null) {
             throw new IllegalArgumentException("register url == null");
         }
         if (logger.isInfoEnabled()) {
             logger.info("Register: " + url);
         }
         registered.add(url);
     }

     @Override
     public void unregister(URL url) {
         if (url == null) {
             throw new IllegalArgumentException("unregister url == null");
         }
         if (logger.isInfoEnabled()) {
             logger.info("Unregister: " + url);
         }
         registered.remove(url);
     }

     @Override
     public void subscribe(URL url, NotifyListener listener) {
         if (url == null) {
             throw new IllegalArgumentException("subscribe url == null");
         }
         if (listener == null) {
             throw new IllegalArgumentException("subscribe listener == null");
         }
         if (logger.isInfoEnabled()) {
             logger.info("Subscribe: " + url);
         }
         Set<NotifyListener> listeners = subscribed.get(url);
         if (listeners == null) {
             subscribed.putIfAbsent(url, new ConcurrentHashSet<NotifyListener>());
             listeners = subscribed.get(url);
         }
         listeners.add(listener);
     }

     @Override
     public void unsubscribe(URL url, NotifyListener listener) {
         if (url == null) {
             throw new IllegalArgumentException("unsubscribe url == null");
         }
         if (listener == null) {
             throw new IllegalArgumentException("unsubscribe listener == null");
         }
         if (logger.isInfoEnabled()) {
             logger.info("Unsubscribe: " + url);
         }
         Set<NotifyListener> listeners = subscribed.get(url);
         if (listeners != null) {
             listeners.remove(listener);
         }
     }

     protected void recover() throws Exception {
         // register
         Set<URL> recoverRegistered = new HashSet<URL>(getRegistered());
         if (!recoverRegistered.isEmpty()) {
             if (logger.isInfoEnabled()) {
                 logger.info("Recover register url " + recoverRegistered);
             }
             for (URL url : recoverRegistered) {
                 register(url);
             }
         }
         // subscribe
         Map<URL, Set<NotifyListener>> recoverSubscribed = new HashMap<URL, Set<NotifyListener>>(getSubscribed());
         if (!recoverSubscribed.isEmpty()) {
             if (logger.isInfoEnabled()) {
                 logger.info("Recover subscribe url " + recoverSubscribed.keySet());
             }
             for (Map.Entry<URL, Set<NotifyListener>> entry : recoverSubscribed.entrySet()) {
                 URL url = entry.getKey();
                 for (NotifyListener listener : entry.getValue()) {
                     subscribe(url, listener);
                 }
             }
         }
     }

     protected void notify(List<URL> urls) {
         if (urls == null || urls.isEmpty()) return;

         for (Map.Entry<URL, Set<NotifyListener>> entry : getSubscribed().entrySet()) {
             URL url = entry.getKey();

             if (!UrlUtils.isMatch(url, urls.get(0))) {
                 continue;
             }

             Set<NotifyListener> listeners = entry.getValue();
             if (listeners != null) {
                 for (NotifyListener listener : listeners) {
                     try {
                         notify(url, listener, filterEmpty(url, urls));
                     } catch (Throwable t) {
                         logger.error("Failed to notify registry event, urls: " + urls + ", cause: " + t.getMessage(), t);
                     }
                 }
             }
         }
     }

     protected void notify(URL url, NotifyListener listener, List<URL> urls) {
         if (url == null) {
             throw new IllegalArgumentException("notify url == null");
         }
         if (listener == null) {
             throw new IllegalArgumentException("notify listener == null");
         }
         if ((urls == null || urls.isEmpty())
                 && !Constants.ANY_VALUE.equals(url.getServiceInterface())) {
             logger.warn("Ignore empty notify urls for subscribe url " + url);
             return;
         }
         if (logger.isInfoEnabled()) {
             logger.info("Notify urls for subscribe url " + url + ", urls: " + urls);
         }
         Map<String, List<URL>> result = new HashMap<String, List<URL>>();
         for (URL u : urls) {
             if (UrlUtils.isMatch(url, u)) {
                 String category = u.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
                 List<URL> categoryList = result.get(category);
                 if (categoryList == null) {
                     categoryList = new ArrayList<URL>();
                     result.put(category, categoryList);
                 }
                 categoryList.add(u);
             }
         }
         if (result.size() == 0) {
             return;
         }
         Map<String, List<URL>> categoryNotified = notified.get(url);
         if (categoryNotified == null) {
             notified.putIfAbsent(url, new ConcurrentHashMap<String, List<URL>>());
             categoryNotified = notified.get(url);
         }
         for (Map.Entry<String, List<URL>> entry : result.entrySet()) {
             String category = entry.getKey();
             List<URL> categoryList = entry.getValue();
             categoryNotified.put(category, categoryList);
             saveProperties(url);
             listener.notify(categoryList);
         }
     }

     private void saveProperties(URL url) {
         if (file == null) {
             return;
         }

         try {
             StringBuilder buf = new StringBuilder();
             Map<String, List<URL>> categoryNotified = notified.get(url);
             if (categoryNotified != null) {
                 for (List<URL> us : categoryNotified.values()) {
                     for (URL u : us) {
                         if (buf.length() > 0) {
                             buf.append(URL_SEPARATOR);
                         }
                         buf.append(u.toFullString());
                     }
                 }
             }
             properties.setProperty(url.getServiceKey(), buf.toString());
             long version = lastCacheChanged.incrementAndGet();
             if (syncSaveFile) {
                 doSaveProperties(version);
             } else {
                 registryCacheExecutor.execute(new SaveProperties(version));
             }
         } catch (Throwable t) {
             logger.warn(t.getMessage(), t);
         }
     }

     @Override
     public void destroy() {
         if (logger.isInfoEnabled()) {
             logger.info("Destroy registry:" + getUrl());
         }
         Set<URL> destroyRegistered = new HashSet<URL>(getRegistered());
         if (!destroyRegistered.isEmpty()) {
             for (URL url : new HashSet<URL>(getRegistered())) {
                 if (url.getParameter(Constants.DYNAMIC_KEY, true)) {
                     try {
                         unregister(url);
                         if (logger.isInfoEnabled()) {
                             logger.info("Destroy unregister url " + url);
                         }
                     } catch (Throwable t) {
                         logger.warn("Failed to unregister url " + url + " to registry " + getUrl() + " on destroy, cause: " + t.getMessage(), t);
                     }
                 }
             }
         }
         Map<URL, Set<NotifyListener>> destroySubscribed = new HashMap<URL, Set<NotifyListener>>(getSubscribed());
         if (!destroySubscribed.isEmpty()) {
             for (Map.Entry<URL, Set<NotifyListener>> entry : destroySubscribed.entrySet()) {
                 URL url = entry.getKey();
                 for (NotifyListener listener : entry.getValue()) {
                     try {
                         unsubscribe(url, listener);
                         if (logger.isInfoEnabled()) {
                             logger.info("Destroy unsubscribe url " + url);
                         }
                     } catch (Throwable t) {
                         logger.warn("Failed to unsubscribe url " + url + " to registry " + getUrl() + " on destroy, cause: " + t.getMessage(), t);
                     }
                 }
             }
         }
     }

     @Override
     public String toString() {
         return getUrl().toString();
     }

     private class SaveProperties implements Runnable {
         private long version;

         private SaveProperties(long version) {
             this.version = version;
         }

         @Override
         public void run() {
             doSaveProperties(version);
         }
     }

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package com.alibaba.dubbo.registry.support;

	import com.alibaba.dubbo.common.Constants;
	import com.alibaba.dubbo.common.URL;
	import com.alibaba.dubbo.common.logger.Logger;
	import com.alibaba.dubbo.common.logger.LoggerFactory;
	import com.alibaba.dubbo.common.utils.ConcurrentHashSet;
	import com.alibaba.dubbo.common.utils.ConfigUtils;
	import com.alibaba.dubbo.common.utils.NamedThreadFactory;
	import com.alibaba.dubbo.common.utils.UrlUtils;
	import com.alibaba.dubbo.registry.NotifyListener;
	import com.alibaba.dubbo.registry.Registry;

	import java.io.File;
	import java.io.FileInputStream;
	import java.io.FileOutputStream;
	import java.io.IOException;
	import java.io.InputStream;
	import java.io.RandomAccessFile;
	import java.nio.channels.FileChannel;
	import java.nio.channels.FileLock;
	import java.util.ArrayList;
	import java.util.HashMap;
	import java.util.HashSet;
	import java.util.List;
	import java.util.Map;
	import java.util.Properties;
	import java.util.Set;
	import java.util.concurrent.ConcurrentHashMap;
	import java.util.concurrent.ConcurrentMap;
	import java.util.concurrent.ExecutorService;
	import java.util.concurrent.Executors;
	import java.util.concurrent.atomic.AtomicLong;
	import java.util.concurrent.atomic.AtomicReference;

	/**
	* AbstractRegistry. (SPI, Prototype, ThreadSafe)
	*
	*/
	public abstract class AbstractRegistry implements Registry {

	// URL address separator, used in file cache, service provider URL separation
	private static final char URL_SEPARATOR = ' ';
	// URL address separated regular expression for parsing the service provider URL list in the file cache
	private static final String URL_SPLIT = "\\s+";
	// Log output
	protected final Logger logger = LoggerFactory.getLogger(getClass());
	// Local disk cache, where the special key value.registies records the list of registry centers, and the others are the list of notified service providers
	private final Properties properties = new Properties();
	// File cache timing writing
	private final ExecutorService registryCacheExecutor = Executors.newFixedThreadPool(1, new NamedThreadFactory("DubboSaveRegistryCache", true));
	// Is it synchronized to save the file
	private final boolean syncSaveFile;
	private final AtomicLong lastCacheChanged = new AtomicLong();
	private final Set<URL> registered = new ConcurrentHashSet<URL>();
	private final ConcurrentMap<URL, Set<NotifyListener>> subscribed = new ConcurrentHashMap<URL, Set<NotifyListener>>();
	private final ConcurrentMap<URL, Map<String, List<URL>>> notified = new ConcurrentHashMap<URL, Map<String, List<URL>>>();
	private URL registryUrl;
	// Local disk cache file
	private File file;

	public AbstractRegistry(URL url) {
	setUrl(url);
	// Start file save timer
	syncSaveFile = url.getParameter(Constants.REGISTRY_FILESAVE_SYNC_KEY, false);
	String filename = url.getParameter(Constants.FILE_KEY, System.getProperty("user.home") + "/.dubbo/dubbo-registry-" + url.getParameter(Constants.APPLICATION_KEY) + "-" + url.getAddress() + ".cache");
	File file = null;
	if (ConfigUtils.isNotEmpty(filename)) {
	file = new File(filename);
	if (!file.exists() && file.getParentFile() != null && !file.getParentFile().exists()) {
	if (!file.getParentFile().mkdirs()) {
	throw new IllegalArgumentException("Invalid registry store file " + file + ", cause: Failed to create directory " + file.getParentFile() + "!");
	}
	}
	}
	this.file = file;
	loadProperties();
	notify(url.getBackupUrls());
	}

	protected static List<URL> filterEmpty(URL url, List<URL> urls) {
	if (urls == null \|\| urls.isEmpty()) {
	List<URL> result = new ArrayList<URL>(1);
	result.add(url.setProtocol(Constants.EMPTY_PROTOCOL));
	return result;
	}
	return urls;
	}

	@Override
	public URL getUrl() {
	return registryUrl;
	}

	protected void setUrl(URL url) {
	if (url == null) {
	throw new IllegalArgumentException("registry url == null");
	}
	this.registryUrl = url;
	}

	public Set<URL> getRegistered() {
	return registered;
	}

	public Map<URL, Set<NotifyListener>> getSubscribed() {
	return subscribed;
	}

	public Map<URL, Map<String, List<URL>>> getNotified() {
	return notified;
	}

	public File getCacheFile() {
	return file;
	}

	public Properties getCacheProperties() {
	return properties;
	}

	public AtomicLong getLastCacheChanged() {
	return lastCacheChanged;
	}

	public void doSaveProperties(long version) {
	if (version < lastCacheChanged.get()) {
	return;
	}
	if (file == null) {
	return;
	}
	// Save
	try {
	File lockfile = new File(file.getAbsolutePath() + ".lock");
	if (!lockfile.exists()) {
	lockfile.createNewFile();
	}
	RandomAccessFile raf = new RandomAccessFile(lockfile, "rw");
	try {
	FileChannel channel = raf.getChannel();
	try {
	FileLock lock = channel.tryLock();
	if (lock == null) {
	throw new IOException("Can not lock the registry cache file " + file.getAbsolutePath() + ", ignore and retry later, maybe multi java process use the file, please config: dubbo.registry.file=xxx.properties");
	}
	// Save
	try {
	if (!file.exists()) {
	file.createNewFile();
	}
	FileOutputStream outputFile = new FileOutputStream(file);
	try {
	properties.store(outputFile, "Dubbo Registry Cache");
	} finally {
	outputFile.close();
	}
	} finally {
	lock.release();
	}
	} finally {
	channel.close();
	}
	} finally {
	raf.close();
	}
	} catch (Throwable e) {
	if (version < lastCacheChanged.get()) {
	return;
	} else {
	registryCacheExecutor.execute(new SaveProperties(lastCacheChanged.incrementAndGet()));
	}
	logger.warn("Failed to save registry store file, cause: " + e.getMessage(), e);
	}
	}

	private void loadProperties() {
	if (file != null && file.exists()) {
	InputStream in = null;
	try {
	in = new FileInputStream(file);
	properties.load(in);
	if (logger.isInfoEnabled()) {
	logger.info("Load registry store file " + file + ", data: " + properties);
	}
	} catch (Throwable e) {
	logger.warn("Failed to load registry store file " + file, e);
	} finally {
	if (in != null) {
	try {
	in.close();
	} catch (IOException e) {
	logger.warn(e.getMessage(), e);
	}
	}
	}
	}
	}

	public List<URL> getCacheUrls(URL url) {
	for (Map.Entry<Object, Object> entry : properties.entrySet()) {
	String key = (String) entry.getKey();
	String value = (String) entry.getValue();
	if (key != null && key.length() > 0 && key.equals(url.getServiceKey())
	&& (Character.isLetter(key.charAt(0)) \|\| key.charAt(0) == '_')
	&& value != null && value.length() > 0) {
	String[] arr = value.trim().split(URL_SPLIT);
	List<URL> urls = new ArrayList<URL>();
	for (String u : arr) {
	urls.add(URL.valueOf(u));
	}
	return urls;
	}
	}
	return null;
	}

	@Override
	public List<URL> lookup(URL url) {
	List<URL> result = new ArrayList<URL>();
	Map<String, List<URL>> notifiedUrls = getNotified().get(url);
	if (notifiedUrls != null && notifiedUrls.size() > 0) {
	for (List<URL> urls : notifiedUrls.values()) {
	for (URL u : urls) {
	if (!Constants.EMPTY_PROTOCOL.equals(u.getProtocol())) {
	result.add(u);
	}
	}
	}
	} else {
	final AtomicReference<List<URL>> reference = new AtomicReference<List<URL>>();
	NotifyListener listener = new NotifyListener() {
	@Override
	public void notify(List<URL> urls) {
	reference.set(urls);
	}
	};
	subscribe(url, listener); // Subscribe logic guarantees the first notify to return
	List<URL> urls = reference.get();
	if (urls != null && !urls.isEmpty()) {
	for (URL u : urls) {
	if (!Constants.EMPTY_PROTOCOL.equals(u.getProtocol())) {
	result.add(u);
	}
	}
	}
	}
	return result;
	}

	@Override
	public void register(URL url) {
	if (url == null) {
	throw new IllegalArgumentException("register url == null");
	}
	if (logger.isInfoEnabled()) {
	logger.info("Register: " + url);
	}
	registered.add(url);
	}

	@Override
	public void unregister(URL url) {
	if (url == null) {
	throw new IllegalArgumentException("unregister url == null");
	}
	if (logger.isInfoEnabled()) {
	logger.info("Unregister: " + url);
	}
	registered.remove(url);
	}

	@Override
	public void subscribe(URL url, NotifyListener listener) {
	if (url == null) {
	throw new IllegalArgumentException("subscribe url == null");
	}
	if (listener == null) {
	throw new IllegalArgumentException("subscribe listener == null");
	}
	if (logger.isInfoEnabled()) {
	logger.info("Subscribe: " + url);
	}
	Set<NotifyListener> listeners = subscribed.get(url);
	if (listeners == null) {
	subscribed.putIfAbsent(url, new ConcurrentHashSet<NotifyListener>());
	listeners = subscribed.get(url);
	}
	listeners.add(listener);
	}

	@Override
	public void unsubscribe(URL url, NotifyListener listener) {
	if (url == null) {
	throw new IllegalArgumentException("unsubscribe url == null");
	}
	if (listener == null) {
	throw new IllegalArgumentException("unsubscribe listener == null");
	}
	if (logger.isInfoEnabled()) {
	logger.info("Unsubscribe: " + url);
	}
	Set<NotifyListener> listeners = subscribed.get(url);
	if (listeners != null) {
	listeners.remove(listener);
	}
	}

	protected void recover() throws Exception {
	// register
	Set<URL> recoverRegistered = new HashSet<URL>(getRegistered());
	if (!recoverRegistered.isEmpty()) {
	if (logger.isInfoEnabled()) {
	logger.info("Recover register url " + recoverRegistered);
	}
	for (URL url : recoverRegistered) {
	register(url);
	}
	}
	// subscribe
	Map<URL, Set<NotifyListener>> recoverSubscribed = new HashMap<URL, Set<NotifyListener>>(getSubscribed());
	if (!recoverSubscribed.isEmpty()) {
	if (logger.isInfoEnabled()) {
	logger.info("Recover subscribe url " + recoverSubscribed.keySet());
	}
	for (Map.Entry<URL, Set<NotifyListener>> entry : recoverSubscribed.entrySet()) {
	URL url = entry.getKey();
	for (NotifyListener listener : entry.getValue()) {
	subscribe(url, listener);
	}
	}
	}
	}

	protected void notify(List<URL> urls) {
	if (urls == null \|\| urls.isEmpty()) return;

	for (Map.Entry<URL, Set<NotifyListener>> entry : getSubscribed().entrySet()) {
	URL url = entry.getKey();

	if (!UrlUtils.isMatch(url, urls.get(0))) {
	continue;
	}

	Set<NotifyListener> listeners = entry.getValue();
	if (listeners != null) {
	for (NotifyListener listener : listeners) {
	try {
	notify(url, listener, filterEmpty(url, urls));
	} catch (Throwable t) {
	logger.error("Failed to notify registry event, urls: " + urls + ", cause: " + t.getMessage(), t);
	}
	}
	}
	}
	}

	protected void notify(URL url, NotifyListener listener, List<URL> urls) {
	if (url == null) {
	throw new IllegalArgumentException("notify url == null");
	}
	if (listener == null) {
	throw new IllegalArgumentException("notify listener == null");
	}
	if ((urls == null \|\| urls.isEmpty())
	&& !Constants.ANY_VALUE.equals(url.getServiceInterface())) {
	logger.warn("Ignore empty notify urls for subscribe url " + url);
	return;
	}
	if (logger.isInfoEnabled()) {
	logger.info("Notify urls for subscribe url " + url + ", urls: " + urls);
	}
	Map<String, List<URL>> result = new HashMap<String, List<URL>>();
	for (URL u : urls) {
	if (UrlUtils.isMatch(url, u)) {
	String category = u.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
	List<URL> categoryList = result.get(category);
	if (categoryList == null) {
	categoryList = new ArrayList<URL>();
	result.put(category, categoryList);
	}
	categoryList.add(u);
	}
	}
	if (result.size() == 0) {
	return;
	}
	Map<String, List<URL>> categoryNotified = notified.get(url);
	if (categoryNotified == null) {
	notified.putIfAbsent(url, new ConcurrentHashMap<String, List<URL>>());
	categoryNotified = notified.get(url);
	}
	for (Map.Entry<String, List<URL>> entry : result.entrySet()) {
	String category = entry.getKey();
	List<URL> categoryList = entry.getValue();
	categoryNotified.put(category, categoryList);
	saveProperties(url);
	listener.notify(categoryList);
	}
	}

	private void saveProperties(URL url) {
	if (file == null) {
	return;
	}

	try {
	StringBuilder buf = new StringBuilder();
	Map<String, List<URL>> categoryNotified = notified.get(url);
	if (categoryNotified != null) {
	for (List<URL> us : categoryNotified.values()) {
	for (URL u : us) {
	if (buf.length() > 0) {
	buf.append(URL_SEPARATOR);
	}
	buf.append(u.toFullString());
	}
	}
	}
	properties.setProperty(url.getServiceKey(), buf.toString());
	long version = lastCacheChanged.incrementAndGet();
	if (syncSaveFile) {
	doSaveProperties(version);
	} else {
	registryCacheExecutor.execute(new SaveProperties(version));
	}
	} catch (Throwable t) {
	logger.warn(t.getMessage(), t);
	}
	}

	@Override
	public void destroy() {
	if (logger.isInfoEnabled()) {
	logger.info("Destroy registry:" + getUrl());
	}
	Set<URL> destroyRegistered = new HashSet<URL>(getRegistered());
	if (!destroyRegistered.isEmpty()) {
	for (URL url : new HashSet<URL>(getRegistered())) {
	if (url.getParameter(Constants.DYNAMIC_KEY, true)) {
	try {
	unregister(url);
	if (logger.isInfoEnabled()) {
	logger.info("Destroy unregister url " + url);
	}
	} catch (Throwable t) {
	logger.warn("Failed to unregister url " + url + " to registry " + getUrl() + " on destroy, cause: " + t.getMessage(), t);
	}
	}
	}
	}
	Map<URL, Set<NotifyListener>> destroySubscribed = new HashMap<URL, Set<NotifyListener>>(getSubscribed());
	if (!destroySubscribed.isEmpty()) {
	for (Map.Entry<URL, Set<NotifyListener>> entry : destroySubscribed.entrySet()) {
	URL url = entry.getKey();
	for (NotifyListener listener : entry.getValue()) {
	try {
	unsubscribe(url, listener);
	if (logger.isInfoEnabled()) {
	logger.info("Destroy unsubscribe url " + url);
	}
	} catch (Throwable t) {
	logger.warn("Failed to unsubscribe url " + url + " to registry " + getUrl() + " on destroy, cause: " + t.getMessage(), t);
	}
	}
	}
	}
	}

	@Override
	public String toString() {
	return getUrl().toString();
	}

	private class SaveProperties implements Runnable {
	private long version;

	private SaveProperties(long version) {
	this.version = version;
	}

	@Override
	public void run() {
	doSaveProperties(version);
	}
	}

	}