Feature/obanyc 2916

onebusaway-nyc-queue-http-proxy/src/main/java/org/onebusaway/nyc/queue_http_proxy/BHSListenerServlet.java

@@ -19,7 +19,6 @@
 import java.io.IOException;
 import java.io.ByteArrayOutputStream;
 import java.net.InetAddress;
-import java.net.UnknownHostException;
 import java.util.TimerTask;
 
 import javax.servlet.ServletContext;
@@ -30,7 +29,7 @@
 import javax.servlet.http.HttpServletResponse;
 
 import org.apache.commons.compress.utils.IOUtils;
-import org.onebusaway.nyc.queue.DNSResolver;
+import org.onebusaway.nyc.queue.LeadershipElectionResolver;
 import org.onebusaway.nyc.queue.IPublisher;
 import org.onebusaway.nyc.queue.Publisher;
 import org.slf4j.Logger;
@@ -56,7 +55,7 @@ public class BHSListenerServlet extends HttpServlet {
 
   private static final long serialVersionUID = 245140554274414196L;
   private static Logger _log = LoggerFactory.getLogger(BHSListenerServlet.class);
-  protected DNSResolver _resolver = null;
+  protected LeadershipElectionResolver _resolver = null;
 
   @Autowired
   private ThreadPoolTaskScheduler _taskScheduler;
@@ -67,17 +66,6 @@ public class BHSListenerServlet extends HttpServlet {
   private static final String DEFAULT_HOST = "*";
   private static final int DEFAULT_PORT = 5563;
 
-  public void startDNSCheckThread() {
-    String host = getHost();
-    _log.info("listening on interface " + host);
-    _resolver = new DNSResolver(host);
-
-    if (_taskScheduler != null) {
-      DNSCheckThread dnsCheckThread = new DNSCheckThread();
-      // ever 10 seconds
-      _taskScheduler.scheduleWithFixedDelay(dnsCheckThread, 10 * 1000);
-    }
-  }
 
   private String getHost() {
     try {
@@ -89,7 +77,6 @@ private String getHost() {
   }
 
   public synchronized void init() throws ServletException {
-    startDNSCheckThread();
 
     IPublisher publisher = (IPublisher) getServletConfig().getServletContext().getAttribute(
         PUBLISHER_KEY);
@@ -185,19 +172,4 @@ private int getInitParameter(String key, int defaultValue) {
     return valueAsInt;
   }
 
-  private class DNSCheckThread extends TimerTask {
-
-    @Override
-    public void run() {
-      try {
-        if (_resolver.hasAddressChanged()) {
-          _log.warn("Resolver Changed -- re-binding queue connection");
-          init();
-        }
-      } catch (Exception e) {
-        _log.error(e.toString());
-        _resolver.reset();
-      }
-    }
-  }
 }

onebusaway-nyc-queue-subscriber/pom.xml

@@ -16,6 +16,11 @@
       <artifactId>onebusaway-nyc-util</artifactId>
       <version>${project.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.onebusaway</groupId>
+      <artifactId>onebusaway-cloud-aws</artifactId>
+      <version>0.0.13-SNAPSHOT</version>
+    </dependency>
     <dependency>
       <groupId>org.zeromq</groupId>
       <artifactId>jeromq</artifactId>

onebusaway-nyc-queue-subscriber/src/main/java/org/onebusaway/nyc/queue/LeadershipElectionResolver.java

@@ -0,0 +1,77 @@
+/**
+ * Copyright (C) 2011 Metropolitan Transportation Authority
+ *
+ * Licensed 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.onebusaway.nyc.queue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+
+import org.onebusaway.cloud.api.ExternalServices;
+import org.onebusaway.cloud.api.ExternalServicesBridgeFactory;
+
+/**
+ * Utility class to test if DNS resolution of a hostname has changed.
+ */
+public class LeadershipElectionResolver {
+	protected static Logger _log = LoggerFactory.getLogger(LeadershipElectionResolver.class);
+	private ExternalServices externalServices = new ExternalServicesBridgeFactory().getExternalServices();
+
+	boolean primaryHasChanged = false;
+	boolean isPrimary;
+
+
+	/**
+	 * Tests if this host is at the IP corresponding to the DNS address.
+	 */
+	public boolean isPrimary() {
+		boolean result = externalServices.isInstancePrimary();
+		primaryHasChanged = (result == isPrimary);
+		isPrimary = result;
+		return isPrimary;
+	}
+
+	/**
+	 * convenience null-safe wrapper for InetAddress.getLocalhost.
+	 */
+	public InetAddress getLocalHost() {
+		try {
+			return InetAddress.getLocalHost();
+		} catch (UnknownHostException uhe) {
+			_log.error(uhe.toString());
+		}
+		return null;
+	}
+
+	/**
+	 * null-safe toString of InetAddress.getLocalhost.
+	 */
+	public String getLocalHostString() {
+		InetAddress local = getLocalHost();
+		if (local != null) {
+			return local.toString();
+		}
+		return "unknown";
+	}
+
+public boolean getPrimaryHasChanged(){
+	return primaryHasChanged;
+	}
+
+
+}

onebusaway-nyc-queue-subscriber/src/main/java/org/onebusaway/nyc/queue/QueueListenerTask.java

@@ -26,7 +26,6 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.module.jaxb.JaxbAnnotationModule;
-import org.onebusaway.nyc.queue.DNSResolver;
 import org.onebusaway.nyc.util.configuration.ConfigurationService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -50,7 +49,7 @@ public abstract class QueueListenerTask {
 	protected boolean _initialized = false;
 	protected ObjectMapper _mapper = new ObjectMapper().registerModule(new JaxbAnnotationModule());
 
-	protected DNSResolver _resolver = null;
+	protected LeadershipElectionResolver _resolver = null;
 	protected ZMQ.Context _context = null;
 	protected ZMQ.Socket _socket = null;
 	protected ZMQ.Poller _poller = null;
@@ -74,16 +73,6 @@ public abstract class QueueListenerTask {
 	public void setCountInterval(int countInterval) {
 	  this._countInterval = countInterval;  
 	}
-
-	public void startDNSCheckThread() {
-		String host = getQueueHost();
-		_resolver = new DNSResolver(host);
-		if (_taskScheduler != null) {
-			DNSCheckThread dnsCheckThread = new DNSCheckThread();
-			// ever 10 seconds
-			_taskScheduler.scheduleWithFixedDelay(dnsCheckThread, 10 * 1000);
-		}
-	}
 
 	private class ReadThread implements Runnable {
 
@@ -151,7 +140,6 @@ public void run() {
 	public void setup() {
 		_executorService = Executors.newFixedThreadPool(1);
 		startListenerThread();
-		startDNSCheckThread();
 		_log.warn("threads started for queue " + getQueueName());
 	}
 
@@ -203,20 +191,5 @@ protected synchronized void initializeQueue(String host, String queueName,
 
 	}
 
-	private class DNSCheckThread extends TimerTask {
-
-		@Override
-		public void run() {
-			try {
-				if (_resolver.hasAddressChanged()) {
-					_log.warn("Resolver Changed -- re-binding queue connection");
-					reinitializeQueue();
-				}
-			} catch (Exception e) {
-				_log.error(e.toString());
-				_resolver.reset();
-			}
-		}
-	}
 
 }

onebusaway-nyc-transit-data-federation/src/main/java/org/onebusaway/nyc/transit_data_federation/impl/queue/ApcQueueListenerTask.java

@@ -129,13 +129,4 @@ public Integer getQueuePort() {
         return _configurationService.getConfigurationValueAsInteger("tds.apcQueueOutputPort", 5576);
     }
 
-    @Override
-    public void startDNSCheckThread() {
-        if (!useApcIfAvailable()) {
-            _log.error("apc integration disabled; DNS check exiting");
-            return;
-        }
-        _log.info("starting DNS check for APC queue " + getQueueName());
-        super.startDNSCheckThread();
-    }
 }

onebusaway-nyc-transit-data-federation/src/main/java/org/onebusaway/nyc/transit_data_federation/impl/queue/TimeQueueListenerTask.java

@@ -115,12 +115,4 @@ public void startListenerThread() {
 		_initialized = true;
 	}
 
-	@Override
-	public void startDNSCheckThread() {
-	  if (!useTimePredictionsIfAvailable()) {
-	    _log.error("time predictions disabled -- exiting");
-	    return;
-	  } 
-	  super.startDNSCheckThread();
-	}
 }