org.apache.ace.agent/src/org/apache/ace/agent/impl/DiscoveryHandlerImpl.java - ace - 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 org.apache.ace.agent.impl;

 import static org.apache.ace.agent.AgentConstants.CONFIG_DISCOVERY_CHECKING;
 import static org.apache.ace.agent.AgentConstants.CONFIG_DISCOVERY_SERVERURLS;
 import static org.apache.ace.agent.AgentConstants.EVENT_AGENT_CONFIG_CHANGED;

 import java.io.IOException;
 import java.net.HttpURLConnection;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.net.URLConnection;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;

 import org.apache.ace.agent.DiscoveryHandler;
 import org.apache.ace.agent.EventListener;

 /**
  * Default {@link DiscoveryHandler} implementation that reads the serverURL(s) from the configuration using key
  * {@link CONFIG_DISCOVERY_SERVERURLS}.
  */
 public class DiscoveryHandlerImpl extends ComponentBase implements DiscoveryHandler, EventListener {

     /*
      * The caching logic in this code is not strictly thread-safe. It does not need to be considering the use-case and
      * the fact that it is eventually correct. Therefore the implementation is optimized for simplicity, performance and
      * garbage reduction in the performance critical path.
      */

     private static class CheckedURL {

         private final URL m_url;
         private final long m_cacheTime;
         private volatile long m_timestamp = 0l;
         private volatile boolean m_available = false;

         public CheckedURL(URL url, long cacheTime) {
             m_url = url;
             m_cacheTime = cacheTime;
         }

         public void setAvailable(boolean value) {
             m_available = value;
             m_timestamp = System.currentTimeMillis();
         }

         public boolean isAvailable() {
             return m_available;
         }

         public boolean isRecentlyChecked() {
             return m_timestamp > (System.currentTimeMillis() - m_cacheTime);
         }

         public URL getURL() {
             return m_url;
         }
     }

     static final String DEFAULT_SERVER_URL = "http://localhost:8080";

     private static final boolean DEFAULT_CHECK_SERVER_URLS = true;
     private static final long DEFAULT_CACHE_MILLISECONDS = 30000;
     private final Map<String, CheckedURL> m_checkedURLs = new HashMap<>();

     private List<String> m_defaultServerURLs;
     private boolean m_defaultCheckURLs;

     private volatile List<String> m_serverURLs;
     private volatile boolean m_checkURLs;

     public DiscoveryHandlerImpl() {
         this(DEFAULT_SERVER_URL, DEFAULT_CHECK_SERVER_URLS);
     }

     DiscoveryHandlerImpl(String serverURL, boolean checkServerURLs) {
         this(new String[] { serverURL }, checkServerURLs);
     }

     DiscoveryHandlerImpl(String[] serverURLs, boolean checkServerURLs) {
         super("discovery");
         m_defaultServerURLs = m_serverURLs = Arrays.asList(serverURLs);
         // ACE-520 - checking URLs only makes sense if there is more than one URL available...
         m_defaultCheckURLs = m_checkURLs = checkServerURLs && m_defaultServerURLs.size() > 1;
     }

     @Override
     protected void onInit() throws Exception {
         String urls = getConfigurationHandler().get(CONFIG_DISCOVERY_SERVERURLS, mergeUrls(m_defaultServerURLs));
         boolean checkServerURLs = getConfigurationHandler().getBoolean(CONFIG_DISCOVERY_CHECKING, DEFAULT_CHECK_SERVER_URLS);
         m_defaultServerURLs = m_serverURLs = splitUrls(urls);
         // ACE-520 - checking URLs only makes sense if there is more than one URL available...
         m_defaultCheckURLs = m_checkURLs = checkServerURLs && m_defaultServerURLs.size() > 1;
         getEventsHandler().addListener(this);
     }

     @Override
     protected void onStop() throws Exception {
         getEventsHandler().removeListener(this);
         m_checkedURLs.clear();
     }

     @Override
     public void handle(String topic, Map<String, String> payload) {
         if (!EVENT_AGENT_CONFIG_CHANGED.equals(topic)) {
             return;
         }

         List<String> serverURLs;
         String urlsValue = payload.get(CONFIG_DISCOVERY_SERVERURLS);
         if (urlsValue != null && !"".equals(urlsValue.trim())) {
             serverURLs = splitUrls(urlsValue);
         }
         else {
             serverURLs = m_defaultServerURLs;
         }

         String checkingValue = payload.get(CONFIG_DISCOVERY_CHECKING);
         boolean checkURLs;
         if (checkingValue != null && !"".equals(checkingValue.trim())) {
             checkURLs = Boolean.parseBoolean(checkingValue);
         }
         else {
             checkURLs = m_defaultCheckURLs;
         }
         // ACE-520 - checking URLs only makes sense if there is more than one URL available...
         checkURLs &= serverURLs.size() > 1;

         List<String> oldServerURLs = m_serverURLs;
         boolean oldCheckURLs = m_checkURLs;

         if (!oldServerURLs.equals(serverURLs) || oldCheckURLs != checkURLs) {
             m_serverURLs = serverURLs;
             m_checkURLs = checkURLs;
             logDebug("Discovery configuration changed: urls: %s, checking: %s", m_serverURLs, m_checkURLs);
             m_checkedURLs.clear();
         }
     }

     List<String> splitUrls(String urlsValue) {
         return Arrays.asList(urlsValue.trim().split("\\s*,\\s*"));
     }

     String mergeUrls(List<String> urls) {
     	StringBuilder sb = new StringBuilder();
     	for (String url : urls) {
     		if (sb.length() > 0) {
     			sb.append(',');
     		}
     		sb.append(url);
     	}
     	return sb.toString();
     }

     /**
      * Returns the first available URL from a the ordered list of the configured server URLs. If the
      * {@link CONFIG_DISCOVERY_CHECKING} flag is set a connection is opened to test whether a serverURL is available
      * before it is returned.
      *
      * @return a (valid) server URL, or <code>null</code> in case no server URL was valid.
      */
     @Override
     public URL getServerUrl() {
         List<String> serverURLs = m_serverURLs; // local reference
         boolean checkURLs = m_checkURLs; // local value

         URL url = null;
         for (String urlValue : serverURLs) {
             url = getURL(urlValue, checkURLs);
             if (url != null) {
                 break;
             }
         }
         if (url == null) {
             logWarning("No valid server URL discovered?!");
         }
         return url;
     }

     private URL getURL(String serverURL, boolean checkURL) {
         try {
             if (!checkURL) {
                 return new URL(serverURL);
             }

             CheckedURL checkedURL = m_checkedURLs.get(serverURL);
             if (checkedURL == null) {
                 checkedURL = new CheckedURL(new URL(serverURL), DEFAULT_CACHE_MILLISECONDS);
                 m_checkedURLs.put(serverURL, checkedURL);
             }
             else {
                 if (checkedURL.isRecentlyChecked()) {
                     if (checkedURL.isAvailable()) {
                         logDebug("Returning cached serverURL: %s", serverURL);
                         return checkedURL.getURL();
                     }
                     else {
                         logDebug("Ignoring blacklisted serverURL: %s", serverURL);
                         return null;
                     }
                 }
             }
             try {
                 tryConnect(checkedURL.m_url);
                 logDebug("Succesfully connected to serverURL: %s", serverURL);
                 checkedURL.setAvailable(true);
                 return checkedURL.getURL();
             }
             catch (IOException e) {
                 logWarning("Blacklisting unavailable serverURL: %s", serverURL);
                 checkedURL.setAvailable(false);
                 return null;
             }

         }
         catch (MalformedURLException e) {
             logWarning("Ignoring invalid/malformed serverURL: %s", serverURL);
             return null;
         }
     }

     private void tryConnect(URL serverURL) throws IOException {
         URLConnection connection = null;
         try {
             connection = getConnectionHandler().getConnection(serverURL);
             connection.connect();
         }
         finally {
             if (connection instanceof HttpURLConnection) {
                 ((HttpURLConnection) connection).disconnect();
             }
         }
     }
 }
	/*
	* 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 org.apache.ace.agent.impl;

	import static org.apache.ace.agent.AgentConstants.CONFIG_DISCOVERY_CHECKING;
	import static org.apache.ace.agent.AgentConstants.CONFIG_DISCOVERY_SERVERURLS;
	import static org.apache.ace.agent.AgentConstants.EVENT_AGENT_CONFIG_CHANGED;

	import java.io.IOException;
	import java.net.HttpURLConnection;
	import java.net.MalformedURLException;
	import java.net.URL;
	import java.net.URLConnection;
	import java.util.Arrays;
	import java.util.HashMap;
	import java.util.List;
	import java.util.Map;

	import org.apache.ace.agent.DiscoveryHandler;
	import org.apache.ace.agent.EventListener;

	/**
	* Default {@link DiscoveryHandler} implementation that reads the serverURL(s) from the configuration using key
	* {@link CONFIG_DISCOVERY_SERVERURLS}.
	*/
	public class DiscoveryHandlerImpl extends ComponentBase implements DiscoveryHandler, EventListener {

	/*
	* The caching logic in this code is not strictly thread-safe. It does not need to be considering the use-case and
	* the fact that it is eventually correct. Therefore the implementation is optimized for simplicity, performance and
	* garbage reduction in the performance critical path.
	*/

	private static class CheckedURL {

	private final URL m_url;
	private final long m_cacheTime;
	private volatile long m_timestamp = 0l;
	private volatile boolean m_available = false;

	public CheckedURL(URL url, long cacheTime) {
	m_url = url;
	m_cacheTime = cacheTime;
	}

	public void setAvailable(boolean value) {
	m_available = value;
	m_timestamp = System.currentTimeMillis();
	}

	public boolean isAvailable() {
	return m_available;
	}

	public boolean isRecentlyChecked() {
	return m_timestamp > (System.currentTimeMillis() - m_cacheTime);
	}

	public URL getURL() {
	return m_url;
	}
	}

	static final String DEFAULT_SERVER_URL = "http://localhost:8080";

	private static final boolean DEFAULT_CHECK_SERVER_URLS = true;
	private static final long DEFAULT_CACHE_MILLISECONDS = 30000;
	private final Map<String, CheckedURL> m_checkedURLs = new HashMap<>();

	private List<String> m_defaultServerURLs;
	private boolean m_defaultCheckURLs;

	private volatile List<String> m_serverURLs;
	private volatile boolean m_checkURLs;

	public DiscoveryHandlerImpl() {
	this(DEFAULT_SERVER_URL, DEFAULT_CHECK_SERVER_URLS);
	}

	DiscoveryHandlerImpl(String serverURL, boolean checkServerURLs) {
	this(new String[] { serverURL }, checkServerURLs);
	}

	DiscoveryHandlerImpl(String[] serverURLs, boolean checkServerURLs) {
	super("discovery");
	m_defaultServerURLs = m_serverURLs = Arrays.asList(serverURLs);
	// ACE-520 - checking URLs only makes sense if there is more than one URL available...
	m_defaultCheckURLs = m_checkURLs = checkServerURLs && m_defaultServerURLs.size() > 1;
	}

	@Override
	protected void onInit() throws Exception {
	String urls = getConfigurationHandler().get(CONFIG_DISCOVERY_SERVERURLS, mergeUrls(m_defaultServerURLs));
	boolean checkServerURLs = getConfigurationHandler().getBoolean(CONFIG_DISCOVERY_CHECKING, DEFAULT_CHECK_SERVER_URLS);
	m_defaultServerURLs = m_serverURLs = splitUrls(urls);
	// ACE-520 - checking URLs only makes sense if there is more than one URL available...
	m_defaultCheckURLs = m_checkURLs = checkServerURLs && m_defaultServerURLs.size() > 1;
	getEventsHandler().addListener(this);
	}

	@Override
	protected void onStop() throws Exception {
	getEventsHandler().removeListener(this);
	m_checkedURLs.clear();
	}

	@Override
	public void handle(String topic, Map<String, String> payload) {
	if (!EVENT_AGENT_CONFIG_CHANGED.equals(topic)) {
	return;
	}

	List<String> serverURLs;
	String urlsValue = payload.get(CONFIG_DISCOVERY_SERVERURLS);
	if (urlsValue != null && !"".equals(urlsValue.trim())) {
	serverURLs = splitUrls(urlsValue);
	}
	else {
	serverURLs = m_defaultServerURLs;
	}

	String checkingValue = payload.get(CONFIG_DISCOVERY_CHECKING);
	boolean checkURLs;
	if (checkingValue != null && !"".equals(checkingValue.trim())) {
	checkURLs = Boolean.parseBoolean(checkingValue);
	}
	else {
	checkURLs = m_defaultCheckURLs;
	}
	// ACE-520 - checking URLs only makes sense if there is more than one URL available...
	checkURLs &= serverURLs.size() > 1;

	List<String> oldServerURLs = m_serverURLs;
	boolean oldCheckURLs = m_checkURLs;

	if (!oldServerURLs.equals(serverURLs) \|\| oldCheckURLs != checkURLs) {
	m_serverURLs = serverURLs;
	m_checkURLs = checkURLs;
	logDebug("Discovery configuration changed: urls: %s, checking: %s", m_serverURLs, m_checkURLs);
	m_checkedURLs.clear();
	}
	}

	List<String> splitUrls(String urlsValue) {
	return Arrays.asList(urlsValue.trim().split("\\s,\\s"));
	}

	String mergeUrls(List<String> urls) {
	StringBuilder sb = new StringBuilder();
	for (String url : urls) {
	if (sb.length() > 0) {
	sb.append(',');
	}
	sb.append(url);
	}
	return sb.toString();
	}

	/**
	* Returns the first available URL from a the ordered list of the configured server URLs. If the
	* {@link CONFIG_DISCOVERY_CHECKING} flag is set a connection is opened to test whether a serverURL is available
	* before it is returned.
	*
	* @return a (valid) server URL, or <code>null</code> in case no server URL was valid.
	*/
	@Override
	public URL getServerUrl() {
	List<String> serverURLs = m_serverURLs; // local reference
	boolean checkURLs = m_checkURLs; // local value

	URL url = null;
	for (String urlValue : serverURLs) {
	url = getURL(urlValue, checkURLs);
	if (url != null) {
	break;
	}
	}
	if (url == null) {
	logWarning("No valid server URL discovered?!");
	}
	return url;
	}

	private URL getURL(String serverURL, boolean checkURL) {
	try {
	if (!checkURL) {
	return new URL(serverURL);
	}

	CheckedURL checkedURL = m_checkedURLs.get(serverURL);
	if (checkedURL == null) {
	checkedURL = new CheckedURL(new URL(serverURL), DEFAULT_CACHE_MILLISECONDS);
	m_checkedURLs.put(serverURL, checkedURL);
	}
	else {
	if (checkedURL.isRecentlyChecked()) {
	if (checkedURL.isAvailable()) {
	logDebug("Returning cached serverURL: %s", serverURL);
	return checkedURL.getURL();
	}
	else {
	logDebug("Ignoring blacklisted serverURL: %s", serverURL);
	return null;
	}
	}
	}
	try {
	tryConnect(checkedURL.m_url);
	logDebug("Succesfully connected to serverURL: %s", serverURL);
	checkedURL.setAvailable(true);
	return checkedURL.getURL();
	}
	catch (IOException e) {
	logWarning("Blacklisting unavailable serverURL: %s", serverURL);
	checkedURL.setAvailable(false);
	return null;
	}

	}
	catch (MalformedURLException e) {
	logWarning("Ignoring invalid/malformed serverURL: %s", serverURL);
	return null;
	}
	}

	private void tryConnect(URL serverURL) throws IOException {
	URLConnection connection = null;
	try {
	connection = getConnectionHandler().getConnection(serverURL);
	connection.connect();
	}
	finally {
	if (connection instanceof HttpURLConnection) {
	((HttpURLConnection) connection).disconnect();
	}
	}
	}
	}