Merge pull request #114 from lloydwatkin/dns-disco

DNS lookup of channel server

Author: martin-hewitt

pom.xml

@@ -4,12 +4,17 @@
     <groupId>org.buddycloud</groupId>
     <artifactId>channelserver</artifactId>
     <packaging>jar</packaging>
-    <version>0.13.3-SNAPSHOT</version>
+    <version>0.14.0</version>
     <name>Buddycloud Java Server</name>
     <scm>
         <url>https://github.com/buddycloud/buddycloud-server-java</url>
     </scm>
     <dependencies>
+        <dependency>
+	    <groupId>org.xbill</groupId>
+	    <artifactId>dnsjava</artifactId>
+	    <version>2.1.6</version>
+        </dependency>
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
@@ -70,7 +75,7 @@
             <groupId>xmlunit</groupId>
             <artifactId>xmlunit</artifactId>
             <version>1.3</version>
-        </dependency>        
+        </dependency>
         <dependency>
             <groupId>xpp3</groupId>
             <artifactId>xpp3</artifactId>

repo/org/xbill/dnsjava/2.1.6/dnsjava-2.1.6.jar

@@ -60,7 +60,7 @@ public void process(IQ packet) throws Exception {
 		try {
 			processPacket(packet);
 		} catch (Exception e) {
-			if (true == packet.getType().toString().equals("result")) return;
+			if (packet.getType().equals(IQ.Type.result) || packet.getType().equals(IQ.Type.error)) return;
 			IQ reply = IQ.createResultIQ(packet);
 			reply.setChildElement(packet.getChildElement().createCopy());
 			reply.setType(Type.error);
@@ -101,7 +101,7 @@ private void processPacket(IQ packet) throws Exception,
 		}
 		logger.debug("Couldn't find processor for packet");
 
-		if (packet.getType() == IQ.Type.set || packet.getType() == IQ.Type.get) {
+		if (packet.getType().equals(IQ.Type.set) || packet.getType().equals(IQ.Type.get)) {
 
 			IQ reply = IQ.createResultIQ(packet);
 			reply.setChildElement(packet.getChildElement().createCopy());

src/main/java/org/buddycloud/channelserver/packetprocessor/iq/IQProcessor.java

@@ -19,6 +19,11 @@
 import org.xmpp.packet.JID;
 import org.xmpp.packet.Packet;
 import org.xmpp.packet.PacketError;
+import org.xbill.DNS.SRVRecord;
+import org.xbill.DNS.Record;
+import org.xbill.DNS.Type;
+import org.xbill.DNS.Lookup;
+import org.xbill.DNS.TextParseException;
 
 public class FederatedQueueManager {
 	private static final Logger logger = Logger
@@ -32,6 +37,8 @@ public class FederatedQueueManager {
 	public static final String IDENTITY_TYPE_CHANNELS = "channels";
 	public static final String BUDDYCLOUD_SERVER = "buddycloud-server";
 
+    public static final String SRV_PREFIX = "_buddycloud-server._tcp.";
+
 	private int id = 1;
 
 	private final ChannelsEngine component;
@@ -51,7 +58,7 @@ public class FederatedQueueManager {
 	public FederatedQueueManager(ChannelsEngine component, String localServer) {
 		this.component = component;
 		this.localServer = localServer;
-		
+
 		nodeMap.start();
 		sentRemotePackets.start();
 	}
@@ -63,9 +70,9 @@ private int getId() {
 	}
 
 	public void process(Packet packet) throws ComponentException {
-		
+
 		logger.debug("Packet payload " + packet.toXML() + " going to federation.");
-		
+
 		String to = packet.getTo().toString();
 
 		String uniqueId = generateUniqueId(packet);
@@ -101,6 +108,7 @@ public void process(Packet packet) throws ComponentException {
 			waitingStanzas.get(to).add(packet);
 			logger.debug("Adding packet to waiting stanza list for " + to
 					+ " (size " + waitingStanzas.get(to).size() + ")");
+            attemptDnsDiscovery(to);
 		} catch (Exception e) {
 			logger.error(e);
 		}
@@ -175,7 +183,7 @@ public void processDiscoItemsResponse(JID from, List<Element> items)
 	public boolean isFederatedDiscoInfoRequest(String packetId) {
 		return remoteServerInfoRequestIds.containsKey(packetId);
 	}
-	
+
 	private void setDiscoveredServer(String server, String handler) {
 		discoveredServers.put(server, handler);
 	}
@@ -199,28 +207,53 @@ public void processDiscoInfoResponse(JID from, String id,
 				sendFederatedRequests(originatingServer);
 			}
 		}
-		
+
 		if (remoteServerItemsToProcess.get(originatingServer) < 1) {
 			if (!discoveredServers.containsKey(originatingServer)) {
-				sendRemoteChannelServerNotFoundErrorResponses(originatingServer);
-				remoteChannelDiscoveryStatus.put(originatingServer,
-						NO_CHANNEL_SERVER);
-				waitingStanzas.remove(originatingServer);
+                sendRemoteChannelServerNotFoundErrorResponses(originatingServer);
+                remoteChannelDiscoveryStatus.put(originatingServer,
+                        NO_CHANNEL_SERVER);
+                waitingStanzas.remove(originatingServer);
 			} else {
 				remoteChannelDiscoveryStatus.put(originatingServer, DISCOVERED);
 			}
 		}
 	}
 
+    private boolean attemptDnsDiscovery(String originatingServer) throws ComponentException {
+        try {
+            String query = SRV_PREFIX + originatingServer;
+            Record[] records = new Lookup(query, Type.SRV).run();
+            if ((null == records) || (0 == records.length)) {
+                logger.debug("No appropriate DNS entry found for " + originatingServer);
+                return false;
+            }
+            SRVRecord record = (SRVRecord) records[0];
+            String targetServer = record.getTarget().toString(true);
+            setDiscoveredServer(originatingServer, targetServer);
+            logger.info("DNS discovery complete for buddycloud server @ "
+                + originatingServer + " (" + targetServer + ")");
+            remoteChannelDiscoveryStatus.put(originatingServer, DISCOVERED);
+            sendFederatedRequests(originatingServer);
+            return true;
+        } catch (TextParseException e) {
+            logger.error(e);
+            return false;
+        }
+    }
+
 	private void sendFederatedRequests(String originatingServer)
 			throws ComponentException {
 		String remoteServer = discoveredServers.get(originatingServer);
 		List<Packet> packetsToSend = waitingStanzas.get(originatingServer);
 		if (packetsToSend == null) {
+            logger.trace("No queued federated packets to send to " + originatingServer);
 			return;
 		}
+        logger.trace("Catching up on federated packet sending to " + originatingServer);
 		for (Packet packet : packetsToSend) {
 			packet.setTo(remoteServer);
+            logger.trace(packet.toString());
 			sendPacket(packet.createCopy());
 		}
 		waitingStanzas.remove(originatingServer);
@@ -261,6 +294,10 @@ public void passResponseToRequester(IQ packet) throws Exception {
 							+ packet.getID() + ")");
 		}
 
+        if (packet.getType().equals(IQ.Type.error) && !remoteChannelDiscoveryStatus.get(packet.getFrom()).equals(DISCOVERED)) {
+            return;
+        }
+
 		String uniqueId = packet.getID();
 		packet.setID(idMap.get(uniqueId));
 		packet.setTo((JID) sentRemotePackets.get(uniqueId));
@@ -270,7 +307,7 @@ public void passResponseToRequester(IQ packet) throws Exception {
 
 		component.sendPacket(packet);
 	}
-	
+
 	public String getRelatedNodeForRemotePacket(IQ packet) {
 		String id = null;
 		if (nodeMap.containsKey(packet.getID())) {