OF-1309 Route based on DomainPairs (#916)

* OF-1309 Route based on DomainPairs

* Fixes found during test

* Fixes found during test II

* Fixes in plugins (Kraken)

* Update minServerVersion/version for Kraken

src/java/org/jivesoftware/openfire/IQRouter.java

@@ -24,6 +24,7 @@ import org.jivesoftware.openfire.interceptor.PacketRejectedException;
 import org.jivesoftware.openfire.privacy.PrivacyList;
 import org.jivesoftware.openfire.privacy.PrivacyListManager;
 import org.jivesoftware.openfire.session.ClientSession;
+import org.jivesoftware.openfire.session.DomainPair;
 import org.jivesoftware.openfire.session.LocalClientSession;
 import org.jivesoftware.openfire.session.Session;
 import org.jivesoftware.openfire.user.UserManager;
@@ -314,7 +315,7 @@ public class IQRouter extends BasicModule {
         try {
             // Check for registered components, services or remote servers
             if (recipientJID != null &&
-                    (routingTable.hasComponentRoute(recipientJID) || routingTable.hasServerRoute(recipientJID))) {
+                    (routingTable.hasComponentRoute(recipientJID) || routingTable.hasServerRoute(new DomainPair(packet.getFrom().getDomain(), recipientJID.getDomain())))) {
                 // A component/service/remote server was found that can handle the Packet
                 routingTable.routePacket(recipientJID, packet, false);
                 return;

src/java/org/jivesoftware/openfire/RoutingTable.java

@@ -16,10 +16,7 @@
 
 package org.jivesoftware.openfire;
 
-import org.jivesoftware.openfire.session.ClientSession;
-import org.jivesoftware.openfire.session.LocalClientSession;
-import org.jivesoftware.openfire.session.LocalOutgoingServerSession;
-import org.jivesoftware.openfire.session.OutgoingServerSession;
+import org.jivesoftware.openfire.session.*;
 import org.xmpp.packet.JID;
 import org.xmpp.packet.Message;
 import org.xmpp.packet.Packet;
@@ -90,7 +87,7 @@ public interface RoutingTable {
      * @param route the address associated to the route.
      * @param destination the outgoing server session.
      */
-    void addServerRoute(JID route, LocalOutgoingServerSession destination);
+    void addServerRoute(DomainPair route, LocalOutgoingServerSession destination);
 
     /**
      * Adds a route to the routing table for the specified internal or external component. <p>
@@ -197,10 +194,10 @@ public interface RoutingTable {
      * as long as a node has a connection to the remote server a true value will be
      * returned.
      *
-     * @param jid JID that specifies the remote server address.
+     * @param pair DomainPair that specifies the local/remote server address.
      * @return true if an outgoing server session exists to the specified remote server.
      */
-    boolean hasServerRoute(JID jid);
+    boolean hasServerRoute(DomainPair pair);
 
     /**
      * Returns true if an internal or external component is hosting the specified address.
@@ -246,7 +243,7 @@ public interface RoutingTable {
      * @param jid the address of the session.
      * @return the outgoing server session associated to the specified XMPP address or null if none was found.
      */
-    OutgoingServerSession getServerRoute(JID jid);
+    OutgoingServerSession getServerRoute(DomainPair pair);
 
     /**
      * Returns a collection with the hostnames of the remote servers that currently may receive
@@ -256,6 +253,7 @@ public interface RoutingTable {
      *         packets sent from this server.
      */
     Collection<String> getServerHostnames();
+    Collection<DomainPair> getServerRoutes();
 
     /**
      * Returns the number of outgoing server sessions hosted in this JVM. When runing inside of
@@ -320,7 +318,7 @@ public interface RoutingTable {
      * @param route the route to remove.
      * @return true if the route was successfully removed.
      */
-    boolean removeServerRoute(JID route);
+    boolean removeServerRoute(DomainPair route);
 
     /**
      * Returns true if a route of a component has been successfully removed. Both internal

src/java/org/jivesoftware/openfire/SessionManager.java

@@ -897,11 +897,20 @@ public class SessionManager extends BasicModule implements ClusterEventListener/
      * OutgoingServerSession an only send packets to the remote server but are not capable of
      * receiving packets from the remote server.
      *
-     * @param hostname the name of the remote server.
+     * @param pair DomainPair describing the local and remote servers.
      * @return a session that was originated from this server to a remote server.
      */
-    public OutgoingServerSession getOutgoingServerSession(String hostname) {
-        return routingTable.getServerRoute(new JID(null, hostname, null, true));
+    public OutgoingServerSession getOutgoingServerSession(DomainPair pair) {
+        return routingTable.getServerRoute(pair);
+    }
+    public List<OutgoingServerSession> getOutgoingServerSessions(String host) {
+        List<OutgoingServerSession> sessions = new LinkedList<>();
+        for (DomainPair pair : routingTable.getServerRoutes()) {
+            if (pair.getRemote().equals(host)) {
+                sessions.add(routingTable.getServerRoute(pair));
+            }
+        }
+        return sessions;
     }
 
     public Collection<ClientSession> getSessions(String username) {
@@ -1058,6 +1067,9 @@ public class SessionManager extends BasicModule implements ClusterEventListener/
     public Collection<String> getOutgoingServers() {
         return routingTable.getServerHostnames();
     }
+    public Collection<DomainPair> getOutgoingDomainPairs() {
+        return routingTable.getServerRoutes();
+    }
 
     /**
      * Broadcasts the given data to all connected sessions. Excellent
@@ -1344,7 +1356,7 @@ public class SessionManager extends BasicModule implements ClusterEventListener/
             // Remove all the hostnames that were registered for this server session
             for (DomainPair domainPair : session.getOutgoingDomainPairs()) {
                 // Remove the route to the session using the hostname
-                server.getRoutingTable().removeServerRoute(new JID(null, domainPair.getRemote(), null, true));
+                server.getRoutingTable().removeServerRoute(domainPair);
             }
         }
     }

src/java/org/jivesoftware/openfire/server/OutgoingSessionPromise.java

@@ -29,6 +29,7 @@ import org.jivesoftware.openfire.interceptor.PacketInterceptor;
 import org.jivesoftware.openfire.interceptor.PacketRejectedException;
 import org.jivesoftware.openfire.session.ClientSession;
 import org.jivesoftware.openfire.session.ConnectionSettings;
+import org.jivesoftware.openfire.session.DomainPair;
 import org.jivesoftware.openfire.session.LocalOutgoingServerSession;
 import org.jivesoftware.openfire.spi.RoutingTableImpl;
 import org.jivesoftware.util.JiveGlobals;
@@ -261,7 +262,7 @@ public class OutgoingSessionPromise implements RoutableChannelHandler {
                 lock.unlock();
             }
             if (created) {
-                if (!routingTable.hasServerRoute(packet.getTo())) {
+                if (!routingTable.hasServerRoute(new DomainPair(packet.getFrom().getDomain(), packet.getTo().getDomain()))) {
                     throw new Exception("Route created but not found!!!");
                 }
                 // A connection to the remote server was created so get the route and send the packet

src/java/org/jivesoftware/openfire/server/RemoteServerManager.java

@@ -27,6 +27,7 @@ import org.jivesoftware.openfire.ConnectionManager;
 import org.jivesoftware.openfire.SessionManager;
 import org.jivesoftware.openfire.server.RemoteServerConfiguration.Permission;
 import org.jivesoftware.openfire.session.ConnectionSettings;
+import org.jivesoftware.openfire.session.DomainPair;
 import org.jivesoftware.openfire.session.Session;
 import org.jivesoftware.util.JiveGlobals;
 import org.jivesoftware.util.cache.Cache;
@@ -90,11 +91,14 @@ public class RemoteServerManager {
         for (Session session : SessionManager.getInstance().getIncomingServerSessions(domain)) {
             session.close();
         }
-        Session session = SessionManager.getInstance().getOutgoingServerSession(domain);
-        if (session != null) {
+        // Can't just lookup a single remote server anymore, so check them all.
+        for (DomainPair domainPair : SessionManager.getInstance().getOutgoingDomainPairs()) {
+            if (domainPair.getRemote().equals(domain)) {
+                Session session = SessionManager.getInstance().getOutgoingServerSession(domainPair);
                 session.close();
             }
         }
+    }
 
     /**
      * Returns true if the remote server with the specified domain can connect to the
@@ -348,9 +352,9 @@ public class RemoteServerManager {
                 }
             }
         }
-        for (String hostname : SessionManager.getInstance().getOutgoingServers()) {
-            if (!canAccess(hostname)) {
-                Session session = SessionManager.getInstance().getOutgoingServerSession(hostname);
+        for (DomainPair domainPair : SessionManager.getInstance().getOutgoingDomainPairs()) {
+            if (!canAccess(domainPair.getRemote())) {
+                Session session = SessionManager.getInstance().getOutgoingServerSession(domainPair);
                 session.close();
             }
         }

src/java/org/jivesoftware/openfire/session/DomainPair.java

@@ -3,9 +3,10 @@ package org.jivesoftware.openfire.session;
 /**
  * Holds a (possibly authenticated) domain pair.
  */
-public class DomainPair {
+public class DomainPair implements java.io.Serializable {
     private final String local;
     private final String remote;
+    private static final long serialVersionUID = 1L;
 
     public DomainPair(String local, String remote) {
         this.local = local;

src/java/org/jivesoftware/openfire/session/LocalOutgoingServerSession.java

@@ -106,6 +106,7 @@ public class LocalOutgoingServerSession extends LocalServerSession implements Ou
      */
     public static boolean authenticateDomain(final String localDomain, final String remoteDomain) {
         final Logger log = LoggerFactory.getLogger( Log.getName() + "[Authenticate local domain: '" + localDomain + "' to remote domain: '" + remoteDomain + "']" );
+        final DomainPair domainPair = new DomainPair(localDomain, remoteDomain);
 
         log.debug( "Start domain authentication ..." );
         if (remoteDomain == null || remoteDomain.length() == 0 || remoteDomain.trim().indexOf(' ') > -1) {
@@ -128,10 +129,16 @@ public class LocalOutgoingServerSession extends LocalServerSession implements Ou
                 log.warn( "Unable to authenticate: a SessionManager instance is not available. This should not occur unless Openfire is starting up or shutting down." );
                 return false;
             }
-            session = sessionManager.getOutgoingServerSession(remoteDomain);
+            session = sessionManager.getOutgoingServerSession(domainPair);
+            if (session != null && session.checkOutgoingDomainPair(localDomain, remoteDomain))
+            {
+                // Do nothing since the domain has already been authenticated.
+                log.debug( "Authentication successful (domain was already authenticated in the pre-existing session)." );
+                return true;
+            }
             if (session != null && !session.isUsingServerDialback() )
             {
-                log.debug( "Dialback was not used for '{}'. This session cannot be re-used.", remoteDomain );
+                log.debug( "Dialback was not used for '{}'. This session cannot be re-used.", domainPair );
                 session = null;
             }
 
@@ -145,7 +152,7 @@ public class LocalOutgoingServerSession extends LocalServerSession implements Ou
                     for ( String otherRemoteDomain : incomingSession.getValidatedDomains() )
                     {
                         // See if there's an outgoing session to any of the (other) domains hosted by the remote domain.
-                        session = sessionManager.getOutgoingServerSession( otherRemoteDomain );
+                        session = sessionManager.getOutgoingServerSession( new DomainPair(localDomain, otherRemoteDomain) );
                         if (session != null)
                         {
                             log.debug( "An outgoing session to a different domain ('{}') hosted on the remote domain was found.", otherRemoteDomain );
@@ -671,14 +678,9 @@ public class LocalOutgoingServerSession extends LocalServerSession implements Ou
 
     @Override
     public void addOutgoingDomainPair(String localDomain, String remoteDomain) {
-        boolean found = false;
-        for (DomainPair domainPair : outgoingDomainPairs) {
-            if (domainPair.getRemote().equals(remoteDomain)) found = true;
-        }
-        outgoingDomainPairs.add(new DomainPair(localDomain, remoteDomain));
-        if (!found) {
-            XMPPServer.getInstance().getRoutingTable().addServerRoute(new JID(null, remoteDomain, null, true), this);
-        }
+        final DomainPair domainPair = new DomainPair(localDomain, remoteDomain);
+        outgoingDomainPairs.add(domainPair);
+        XMPPServer.getInstance().getRoutingTable().addServerRoute(domainPair, this);
     }
 
     @Override

src/java/org/jivesoftware/openfire/session/RemoteSessionLocator.java

@@ -87,5 +87,5 @@ public interface RemoteSessionLocator {
      * @param address the address that uniquely identifies the session.
      * @return a session surrogate of an incoming server session hosted by a remote cluster node.
      */
-    OutgoingServerSession getOutgoingServerSession(byte[] nodeID, JID address);
+    OutgoingServerSession getOutgoingServerSession(byte[] nodeID, DomainPair address);
 }

src/java/org/jivesoftware/openfire/spi/LocalRoutingTable.java

@@ -25,11 +25,7 @@ import java.util.concurrent.ConcurrentHashMap;
 
 import org.jivesoftware.openfire.RoutableChannelHandler;
 import org.jivesoftware.openfire.SessionManager;
-import org.jivesoftware.openfire.session.LocalClientSession;
-import org.jivesoftware.openfire.session.LocalOutgoingServerSession;
-import org.jivesoftware.openfire.session.LocalSession;
-import org.jivesoftware.openfire.session.OutgoingServerSession;
-import org.jivesoftware.openfire.session.Session;
+import org.jivesoftware.openfire.session.*;
 import org.jivesoftware.util.LocaleUtils;
 import org.jivesoftware.util.TaskEngine;
 import org.slf4j.Logger;
@@ -48,27 +44,30 @@ class LocalRoutingTable {
 	
 	private static final Logger Log = LoggerFactory.getLogger(LocalRoutingTable.class);
 
-    Map<String, RoutableChannelHandler> routes = new ConcurrentHashMap<>();
+    Map<DomainPair, RoutableChannelHandler> routes = new ConcurrentHashMap<>();
 
     /**
      * Adds a route of a local {@link RoutableChannelHandler}
      *
-     * @param address the string representation of the JID associated to the route.
+     * @param pair DomainPair associated to the route.
      * @param route the route hosted by this node.
      * @return true if the element was added or false if was already present.
      */
-    boolean addRoute(String address, RoutableChannelHandler route) {
-        return routes.put(address, route) != route;
+    boolean addRoute(DomainPair pair, RoutableChannelHandler route) {
+        return routes.put(pair, route) != route;
     }
 
     /**
      * Returns the route hosted by this node that is associated to the specified address.
      *
-     * @param address the string representation of the JID associated to the route.
+     * @param pair DomainPair associated to the route.
      * @return the route hosted by this node that is associated to the specified address.
      */
-    RoutableChannelHandler getRoute(String address) {
-        return routes.get(address);
+    RoutableChannelHandler getRoute(DomainPair pair) {
+        return routes.get(pair);
+    }
+    RoutableChannelHandler getRoute(JID jid) {
+        return routes.get(new DomainPair("", jid.toString()));
     }
 
     /**
@@ -119,10 +118,10 @@ class LocalRoutingTable {
     /**
      * Removes a route of a local {@link RoutableChannelHandler}
      *
-     * @param address the string representation of the JID associated to the route.
+     * @param pair DomainPair associated to the route.
      */
-    void removeRoute(String address) {
-        routes.remove(address);
+    void removeRoute(DomainPair pair) {
+        routes.remove(pair);
     }
 
     public void start() {
@@ -154,8 +153,11 @@ class LocalRoutingTable {
         }
     }
 
+    public boolean isLocalRoute(DomainPair pair) {
+        return routes.containsKey(pair);
+    }
     public boolean isLocalRoute(JID jid) {
-        return routes.containsKey(jid.toString());
+        return routes.containsKey(new DomainPair("", jid.toString()));
     }
 
     /**

src/java/org/jivesoftware/util/S2STestService.java

@@ -9,6 +9,7 @@ import org.jivesoftware.openfire.interceptor.InterceptorManager;
 import org.jivesoftware.openfire.interceptor.PacketInterceptor;
 import org.jivesoftware.openfire.interceptor.PacketRejectedException;
 import org.jivesoftware.openfire.server.RemoteServerManager;
+import org.jivesoftware.openfire.session.DomainPair;
 import org.jivesoftware.openfire.session.OutgoingServerSession;
 import org.jivesoftware.openfire.session.Session;
 import org.jivesoftware.util.cert.SANCertificateIdentityMapping;
@@ -56,6 +57,7 @@ public class S2STestService {
     public Map<String, String> run() throws Exception {
         waitUntil = new Semaphore(0);
         Map<String, String> results = new HashMap<>();
+        final DomainPair pair = new DomainPair(XMPPServer.getInstance().getServerInfo().getXMPPDomain(), domain);
 
         // Tear down existing routes.
         final SessionManager sessionManager = SessionManager.getInstance();
@@ -64,7 +66,7 @@ public class S2STestService {
             incomingServerSession.close();
         }
 
-        final Session outgoingServerSession = sessionManager.getOutgoingServerSession( domain );
+        final Session outgoingServerSession = sessionManager.getOutgoingServerSession( pair );
         if ( outgoingServerSession != null )
         {
             outgoingServerSession.close();
@@ -72,7 +74,7 @@ public class S2STestService {
 
         final IQ pingRequest = new IQ( Type.get );
         pingRequest.setChildElement( "ping", IQPingHandler.NAMESPACE );
-        pingRequest.setFrom( XMPPServer.getInstance().getServerInfo().getXMPPDomain() );
+        pingRequest.setFrom( pair.getLocal() );
         pingRequest.setTo( domain );
 
         // Intercept logging.
@@ -145,7 +147,8 @@ public class S2STestService {
      * Logs the status of the session.
      */
     private void logSessionStatus() {
-        OutgoingServerSession session = XMPPServer.getInstance().getSessionManager().getOutgoingServerSession(domain);
+        final DomainPair pair = new DomainPair(XMPPServer.getInstance().getServerInfo().getXMPPDomain(), domain);
+        OutgoingServerSession session = XMPPServer.getInstance().getSessionManager().getOutgoingServerSession(pair);
         if (session != null) {
             int connectionStatus = session.getStatus();
             switch(connectionStatus) {
@@ -168,7 +171,8 @@ public class S2STestService {
      * @return A String representation of the certificate chain for the connection to the domain under test.
      */
     private String getCertificates() {
-        Session session = XMPPServer.getInstance().getSessionManager().getOutgoingServerSession(domain);
+        final DomainPair pair = new DomainPair(XMPPServer.getInstance().getServerInfo().getXMPPDomain(), domain);
+        Session session = XMPPServer.getInstance().getSessionManager().getOutgoingServerSession(pair);
         StringBuilder certs = new StringBuilder();
         if (session != null) {
             Log.info("Successfully negotiated TLS connection.");

src/plugins/hazelcast/src/java/org/jivesoftware/openfire/plugin/session/DeliverRawTextTask.java

@@ -19,6 +19,7 @@ package org.jivesoftware.openfire.plugin.session;
 import org.jivesoftware.openfire.SessionManager;
 import org.jivesoftware.openfire.StreamID;
 import org.jivesoftware.openfire.XMPPServer;
+import org.jivesoftware.openfire.session.DomainPair;
 import org.jivesoftware.openfire.session.Session;
 import org.jivesoftware.openfire.spi.BasicStreamIDFactory;
 import org.jivesoftware.util.Log;
@@ -50,6 +51,7 @@ public class DeliverRawTextTask implements ClusterTask<Void> {
             this.sessionType = SessionType.client;
         }
         else if (remoteSession instanceof RemoteOutgoingServerSession) {
+            Log.error("OutgoingServerSession used with DeliverRawTextTask; should be using DeliverRawTextServerTask: " + remoteSession);
             this.sessionType = SessionType.outgoingServer;
         }
         else if (remoteSession instanceof RemoteComponentSession) {
@@ -114,7 +116,8 @@ public class DeliverRawTextTask implements ClusterTask<Void> {
             return SessionManager.getInstance().getConnectionMultiplexerSession(address);
         }
         else if (sessionType == SessionType.outgoingServer) {
-            return SessionManager.getInstance().getOutgoingServerSession(address.getDomain());
+            Log.error("Trying to write raw data to a server session across the cluster: " + address.toString());
+            return null;
         }
         else if (sessionType == SessionType.incomingServer) {
             return SessionManager.getInstance().getIncomingServerSession(streamID);

src/plugins/hazelcast/src/java/org/jivesoftware/openfire/plugin/session/OutgoingServerSessionTask.java

@@ -17,6 +17,7 @@
 package org.jivesoftware.openfire.plugin.session;
 
 import org.jivesoftware.openfire.SessionManager;
+import org.jivesoftware.openfire.session.DomainPair;
 import org.jivesoftware.openfire.session.OutgoingServerSession;
 import org.jivesoftware.openfire.session.Session;
 import org.jivesoftware.util.cache.ExternalizableUtil;
@@ -33,18 +34,18 @@ import java.io.ObjectOutput;
  * @author Gaston Dombiak
  */
 public class OutgoingServerSessionTask extends RemoteSessionTask {
-    private JID address;
+    private DomainPair address;
 
     public OutgoingServerSessionTask() {
     }
 
-    protected OutgoingServerSessionTask(JID address, Operation operation) {
+    protected OutgoingServerSessionTask(DomainPair address, Operation operation) {
         super(operation);
         this.address = address;
     }
 
     Session getSession() {
-        return SessionManager.getInstance().getOutgoingServerSession(address.getDomain());
+        return SessionManager.getInstance().getOutgoingServerSession(address);
     }
 
     public void run() {
@@ -64,7 +65,7 @@ public class OutgoingServerSessionTask extends RemoteSessionTask {
 
     public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
         super.readExternal(in);
-        address = (JID) ExternalizableUtil.getInstance().readSerializable(in);
+        address = (DomainPair) ExternalizableUtil.getInstance().readSerializable(in);
     }
 
     public String toString() {

src/plugins/hazelcast/src/java/org/jivesoftware/openfire/plugin/session/ProcessPacketTask.java

@@ -21,6 +21,7 @@ import org.dom4j.tree.DefaultElement;
 import org.jivesoftware.openfire.SessionManager;
 import org.jivesoftware.openfire.StreamID;
 import org.jivesoftware.openfire.XMPPServer;
+import org.jivesoftware.openfire.session.DomainPair;
 import org.jivesoftware.openfire.session.Session;
 import org.jivesoftware.openfire.spi.BasicStreamIDFactory;
 import org.jivesoftware.util.Log;
@@ -135,7 +136,8 @@ public class ProcessPacketTask implements ClusterTask<Void> {
             return SessionManager.getInstance().getConnectionMultiplexerSession(address);
         }
         else if (sessionType == SessionType.outgoingServer) {
-            return SessionManager.getInstance().getOutgoingServerSession(address.getDomain());
+            final DomainPair pair = new DomainPair(packet.getFrom().getDomain(), address.getDomain());
+            return SessionManager.getInstance().getOutgoingServerSession(pair);
         }
         else if (sessionType == SessionType.incomingServer) {
             return SessionManager.getInstance().getIncomingServerSession(streamID);

src/plugins/hazelcast/src/java/org/jivesoftware/openfire/plugin/session/RemoteOutgoingServerSession.java

@@ -36,9 +36,11 @@ import java.util.Collection;
 public class RemoteOutgoingServerSession extends RemoteSession implements OutgoingServerSession {
 
     private long usingServerDialback = -1;
+    private final DomainPair pair;
 
-    public RemoteOutgoingServerSession(byte[] nodeID, JID address) {
-        super(nodeID, address);
+    public RemoteOutgoingServerSession(byte[] nodeID, DomainPair address) {
+        super(nodeID, new JID(null, address.getRemote(), null, true));
+        this.pair = address;
     }
 
     public Collection<DomainPair> getOutgoingDomainPairs()
@@ -49,11 +51,11 @@ public class RemoteOutgoingServerSession extends RemoteSession implements Outgoi
 
     public void addOutgoingDomainPair( String local, String remote )
     {
-        doClusterTask(new AddOutgoingDomainPair(address, local, remote ));
+        doClusterTask(new AddOutgoingDomainPair(pair, local, remote ));
     }
 
     public boolean authenticateSubdomain(String domain, String hostname) {
-        ClusterTask task = new AuthenticateSubdomainTask(address, domain, hostname);
+        ClusterTask task = new AuthenticateSubdomainTask(pair, domain, hostname);
         return (Boolean) doSynchronousClusterTask(task);
     }
 
@@ -66,22 +68,49 @@ public class RemoteOutgoingServerSession extends RemoteSession implements Outgoi
     }
 
     public boolean checkOutgoingDomainPair(String localDomain, String remoteDomain) {
-        ClusterTask task = new CheckOutgoingDomainPairTask(address, localDomain, remoteDomain);
+        ClusterTask task = new CheckOutgoingDomainPairTask(pair, localDomain, remoteDomain);
         return (Boolean)doSynchronousClusterTask(task);
     }
 
     RemoteSessionTask getRemoteSessionTask(RemoteSessionTask.Operation operation) {
-        return new OutgoingServerSessionTask(address, operation);
+        return new OutgoingServerSessionTask(pair, operation);
     }
 
     ClusterTask getDeliverRawTextTask(String text) {
-        return new DeliverRawTextTask(this, address, text);
+        return new DeliverRawTextServerTask(pair, text);
     }
 
     ClusterTask getProcessPacketTask(Packet packet) {
         return new ProcessPacketTask(this, address, packet);
     }
 
+    private static class DeliverRawTextServerTask extends OutgoingServerSessionTask {
+        private String text;
+
+        public DeliverRawTextServerTask() {
+            super();
+        }
+
+        protected DeliverRawTextServerTask(DomainPair address, String text) {
+            super(address, null);
+            this.text = text;
+        }
+
+        public void run() {
+            getSession().deliverRawText(text);
+        }
+
+        public void writeExternal(ObjectOutput out) throws IOException {
+            super.writeExternal(out);
+            ExternalizableUtil.getInstance().writeSafeUTF(out, text);
+        }
+
+        public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+            super.readExternal(in);
+            text = ExternalizableUtil.getInstance().readSafeUTF(in);
+        }
+    }
+
     private static class AddOutgoingDomainPair extends OutgoingServerSessionTask {
         private String local;
         private String remote;
@@ -90,7 +119,7 @@ public class RemoteOutgoingServerSession extends RemoteSession implements Outgoi
             super();
         }
 
-        protected AddOutgoingDomainPair(JID address, String local, String remote) {
+        protected AddOutgoingDomainPair(DomainPair address, String local, String remote) {
             super(address, null);
             this.local = local;
             this.remote = remote;
@@ -121,7 +150,7 @@ public class RemoteOutgoingServerSession extends RemoteSession implements Outgoi
             super();
         }
 
-        protected AuthenticateSubdomainTask(JID address, String domain, String hostname) {
+        protected AuthenticateSubdomainTask(DomainPair address, String domain, String hostname) {
             super(address, null);
             this.domain = domain;
             this.hostname = hostname;
@@ -152,7 +181,7 @@ public class RemoteOutgoingServerSession extends RemoteSession implements Outgoi
             super();
         }
 
-        protected CheckOutgoingDomainPairTask(JID address, String local, String remote) {
+        protected CheckOutgoingDomainPairTask(DomainPair address, String local, String remote) {
             super(address, null);
             this.local = local;
             this.remote = remote;

src/plugins/hazelcast/src/java/org/jivesoftware/openfire/plugin/session/RemoteSessionLocator.java

@@ -45,7 +45,7 @@ public class RemoteSessionLocator implements org.jivesoftware.openfire.session.R
         return new RemoteIncomingServerSession(nodeID, streamID);
     }
 
-    public OutgoingServerSession getOutgoingServerSession(byte[] nodeID, JID address) {
+    public OutgoingServerSession getOutgoingServerSession(byte[] nodeID, DomainPair address) {
         return new RemoteOutgoingServerSession(nodeID, address);
     }
 }

src/plugins/hazelcast/src/java/org/jivesoftware/openfire/plugin/util/cache/ClusterListener.java

@@ -34,6 +34,7 @@ import org.jivesoftware.openfire.handler.DirectedPresence;
 import org.jivesoftware.openfire.handler.PresenceUpdateHandler;
 import org.jivesoftware.openfire.plugin.util.cluster.HazelcastClusterNodeInfo;
 import org.jivesoftware.openfire.session.ClientSessionInfo;
+import org.jivesoftware.openfire.session.DomainPair;
 import org.jivesoftware.openfire.session.IncomingServerSession;
 import org.jivesoftware.openfire.session.RemoteSessionLocator;
 import org.jivesoftware.openfire.spi.BasicStreamIDFactory;
@@ -71,20 +72,19 @@ public class ClusterListener implements MembershipListener, LifecycleListener {
 
     private static final int C2S_CACHE_IDX = 0;
     private static final int ANONYMOUS_C2S_CACHE_IDX = 1;
-    private static final int S2S_CACHE_NAME_IDX= 2;
-    private static final int COMPONENT_CACHE_IDX= 3;
+    private static final int COMPONENT_CACHE_IDX= 2;
 
-    private static final int SESSION_INFO_CACHE_IDX = 4;
-    private static final int COMPONENT_SESSION_CACHE_IDX = 5;
-    private static final int CM_CACHE_IDX = 6;
-    private static final int ISS_CACHE_IDX = 7;
+    private static final int SESSION_INFO_CACHE_IDX = 3;
+    private static final int COMPONENT_SESSION_CACHE_IDX = 4;
+    private static final int CM_CACHE_IDX = 5;
+    private static final int ISS_CACHE_IDX = 6;
 
     /**
      * Caches stored in RoutingTable
      */
     Cache<String, ClientRoute> C2SCache;
     Cache<String, ClientRoute> anonymousC2SCache;
-    Cache<String, byte[]> S2SCache;
+    Cache<DomainPair, byte[]> S2SCache;
     Cache<String, Set<NodeID>> componentsCache;
 
     /**
@@ -101,6 +101,7 @@ public class ClusterListener implements MembershipListener, LifecycleListener {
     Cache<String, Collection<DirectedPresence>> directedPresencesCache;
 
     private Map<NodeID, Set<String>[]> nodeSessions = new ConcurrentHashMap<NodeID, Set<String>[]>();
+    private Map<NodeID, Set<DomainPair>> nodeRoutes = new ConcurrentHashMap<>();
     private Map<NodeID, Map<String, Collection<String>>> nodePresences = new ConcurrentHashMap<NodeID, Map<String, Collection<String>>>();
     private boolean seniorClusterMember = CacheFactory.isSeniorClusterMember();
 
@@ -179,9 +180,6 @@ public class ClusterListener implements MembershipListener, LifecycleListener {
         else if (cacheName.equals(anonymousC2SCache.getName())) {
             return allLists[ANONYMOUS_C2S_CACHE_IDX];
         }
-        else if (cacheName.equals(S2SCache.getName())) {
-            return allLists[S2S_CACHE_NAME_IDX];
-        }
         else if (cacheName.equals(componentsCache.getName())) {
             return allLists[COMPONENT_CACHE_IDX];
         }
@@ -210,7 +208,6 @@ public class ClusterListener implements MembershipListener, LifecycleListener {
             new HashSet<String>(),
             new HashSet<String>(),
             new HashSet<String>(),
-            new HashSet<String>(),
             new HashSet<String>()
         };
         nodeSessions.put(nodeKey, allLists);
@@ -282,13 +279,13 @@ public class ClusterListener implements MembershipListener, LifecycleListener {
         }
 
         // Remove outgoing server sessions hosted in node that left the cluster
-        Set<String> remoteServers = lookupJIDList(key, S2SCache.getName());
+        Set<DomainPair> remoteServers = nodeRoutes.get(key);
         if (!remoteServers.isEmpty()) {
-            for (String fullJID : new ArrayList<String>(remoteServers)) {
-                JID serverJID = new JID(fullJID);
-                routingTable.removeServerRoute(serverJID);
+            for (DomainPair domainPair : remoteServers) {
+                routingTable.removeServerRoute(domainPair);
             }
         }
+        nodeRoutes.remove(key);
 
         Set<String> components = lookupJIDList(key, componentsCache.getName());
         if (!components.isEmpty()) {
@@ -585,7 +582,7 @@ public class ClusterListener implements MembershipListener, LifecycleListener {
         ClusterManager.fireJoinedCluster(false);
         addEntryListener(C2SCache, new CacheListener(this, C2SCache.getName()));
         addEntryListener(anonymousC2SCache, new CacheListener(this, anonymousC2SCache.getName()));
-        addEntryListener(S2SCache, new CacheListener(this, S2SCache.getName()));
+        addEntryListener(S2SCache, new S2SCacheListener());
         addEntryListener(componentsCache, new ComponentCacheListener());
 
         addEntryListener(sessionInfoCache, new CacheListener(this, sessionInfoCache.getName()));
@@ -706,4 +703,65 @@ public class ClusterListener implements MembershipListener, LifecycleListener {
         clusterNodesInfo.put(event.getMember().getUuid(), 
         		new HazelcastClusterNodeInfo(event.getMember(), priorNodeInfo.getJoinedTime()));
 	}
+
+    class S2SCacheListener implements EntryListener {
+        public S2SCacheListener() {
+        }
+
+        public void entryAdded(EntryEvent event) {
+            handleEntryEvent(event, false);
+        }
+
+        public void entryUpdated(EntryEvent event) {
+            handleEntryEvent(event, false);
+        }
+
+        public void entryRemoved(EntryEvent event) {
+            handleEntryEvent(event, true);
+        }
+
+        public void entryEvicted(EntryEvent event) {
+            handleEntryEvent(event, true);
+        }
+
+        private void handleEntryEvent(EntryEvent event, boolean removal) {
+            NodeID nodeID = NodeID.getInstance(StringUtils.getBytes(event.getMember().getUuid()));
+            // ignore events which were triggered by this node
+            if (!XMPPServer.getInstance().getNodeID().equals(nodeID)) {
+                Set<DomainPair> sessionJIDS = nodeRoutes.get(nodeID);
+                if (sessionJIDS == null) {
+                    sessionJIDS = new HashSet<>();
+                }
+                if (removal) {
+                    sessionJIDS.remove(event.getKey());
+                }
+                else {
+                    sessionJIDS.add((DomainPair)event.getKey());
+                }
+            }
+        }
+
+        private void handleMapEvent(MapEvent event) {
+            NodeID nodeID = NodeID.getInstance(StringUtils.getBytes(event.getMember().getUuid()));
+            // ignore events which were triggered by this node
+            if (!XMPPServer.getInstance().getNodeID().equals(nodeID)) {
+                Set<DomainPair> sessionJIDS = nodeRoutes.get(nodeID);
+                if (sessionJIDS != null) {
+                    sessionJIDS.clear();
+                }
+            }
+        }
+
+        @Override
+        public void mapCleared(MapEvent event) {
+            handleMapEvent(event);
+        }
+
+        @Override
+        public void mapEvicted(MapEvent event) {
+            handleMapEvent(event);
+        }
+
+    }
+
 }

src/plugins/kraken/plugin.xml

@@ -8,9 +8,9 @@
     <name>Kraken IM Gateway</name>
     <description>Adds transports to other IM networks.</description>
     <author>Daniel Henninger</author>
-    <version>1.3.0</version>
-    <date>10/12/2015</date>
-    <minServerVersion>4.0.0</minServerVersion>
+    <version>1.3.1</version>
+    <date>11/14/2017</date>
+    <minServerVersion>4.2.0</minServerVersion>
     <databaseKey>gateway</databaseKey>
     <databaseVersion>12</databaseVersion>
     <licenseType>apache2</licenseType>

src/plugins/kraken/src/java/net/sf/kraken/TransportInstance.java

@@ -15,12 +15,14 @@ import org.apache.log4j.Logger;
 import org.jivesoftware.openfire.SessionManager;
 import org.jivesoftware.openfire.XMPPServer;
 import org.jivesoftware.openfire.session.Session;
+import org.jivesoftware.openfire.session.OutgoingServerSession;
 import org.jivesoftware.util.JiveGlobals;
 import org.jivesoftware.util.PropertyEventDispatcher;
 import org.jivesoftware.util.PropertyEventListener;
 import org.xmpp.component.ComponentManager;
 
 import java.util.Map;
+import java.util.List;
 
 /**
  * Transport Instance
@@ -157,12 +159,14 @@ public class TransportInstance<B extends TransportBuddy> implements PropertyEven
         }
 
         try {
-            Session sess = sessionManager.getOutgoingServerSession(fullJID);
+            List<OutgoingServerSession> sessions = sessionManager.getOutgoingServerSessions(fullJID);
+            for (OutgoingServerSession sess : sessions) {
                 if (sess != null) {
                     sess.close();
                     pause = true;
                 }
             }
+        }
         catch (Exception ignored) {
             // Session might have disappeared on its own
         }

src/web/server-session-details.jsp

@@ -26,6 +26,7 @@
 <%@ page import="java.util.Calendar" %>
 <%@ page import="java.util.Date" %>
 <%@ page import="java.util.List" %>
+<%@ page import="java.util.Collection" %>
 
 <%@ taglib uri="http://java.sun.com/jsp/jstl/core" prefix="c" %>
 <%@ taglib uri="http://java.sun.com/jsp/jstl/fmt" prefix="fmt" %>
@@ -46,7 +47,7 @@
     // Get the session & address objects
     SessionManager sessionManager = webManager.getSessionManager();
     List<IncomingServerSession> inSessions = sessionManager.getIncomingServerSessions(hostname);
-    OutgoingServerSession outSession = sessionManager.getOutgoingServerSession(hostname);
+    List<OutgoingServerSession> outSessions = sessionManager.getOutgoingServerSessions(hostname);
 
     // Number dateFormatter for all numbers on this page:
     NumberFormat numFormatter = NumberFormat.getNumberInstance();
@@ -81,10 +82,10 @@
             <fmt:message key="server.session.label.connection" />
         </td>
         <td>
-        <% if (!inSessions.isEmpty() && outSession == null) { %>
+        <% if (!inSessions.isEmpty() && outSessions.isEmpty()) { %>
             <img src="images/incoming_32x16.gif" width="32" height="16" border="0" title="<fmt:message key='server.session.connection.incoming' />" alt="<fmt:message key='server.session.connection.incoming' />">
             <fmt:message key="server.session.connection.incoming" />
-        <% } else if (inSessions.isEmpty() && outSession != null) { %>
+        <% } else if (inSessions.isEmpty() && !outSessions.isEmpty()) { %>
             <img src="images/outgoing_32x16.gif" width="32" height="16" border="0" title="<fmt:message key='server.session.connection.outgoing' />" alt="<fmt:message key='server.session.connection.outgoing' />">
             <fmt:message key="server.session.connection.outgoing" />
         <% } else { %>
@@ -103,10 +104,10 @@
                 <%= inSessions.get(0).getHostAddress() %>
                 /
                 <%= inSessions.get(0).getHostName() %>
-	        <% } else if (outSession != null) { %>
-	            <%= outSession.getHostAddress() %>
+	        <% } else if (!outSessions.isEmpty()) { %>
+	            <%= outSessions.get(0).getHostAddress() %>
 	            /
-	            <%= outSession.getHostName() %>
+	            <%= outSessions.get(0).getHostName() %>
 	        <% }
 	       } catch (java.net.UnknownHostException e) { %>
                 Invalid session/connection
@@ -170,8 +171,8 @@
     <br>
 <%  } %>
 
-<%  // Show details of the incoming session
-    if (outSession != null) {
+<%  // Show details of the outgoing sessiona
+    for (OutgoingServerSession outSession : outSessions) {
 %>
     <b><fmt:message key="server.session.details.outgoing_session" /></b>
     <div class="jive-table">

src/web/server-session-row.jspf

@@ -28,8 +28,13 @@
         }
     }
     // Check if outgoing session is secured (only if incoming sessions are secured)
-    if (isSecured && outSession != null) {
-        isSecured = outSession.isSecure();
+    if (isSecured) {
+        for (org.jivesoftware.openfire.session.OutgoingServerSession outSession : outSessions) {
+            if (!outSession.isSecure()) {
+                isSecured = false;
+                break;
+            }
+        }
     }
 %>
 <tr class="jive-<%= (((count % 2) == 0) ? "even" : "odd") %>">
@@ -49,12 +54,12 @@
      <% } else { %>
     <td width="1%"><img src="images/blank.gif" width="1" height="1" alt=""></td>
      <% } %>
-    <% if (!inSessions.isEmpty() && outSession == null) { %>
+    <% if (!inSessions.isEmpty() && outSessions.isEmpty()) { %>
         <td width="1%">
             <img src="images/incoming_32x16.gif" width="32" height="16" border="0" title="<fmt:message key='server.session.connection.incoming' />" alt="<fmt:message key='server.session.connection.incoming' />">
         </td>
         <td width="10%"><fmt:message key="server.session.connection.incoming" /></td>
-    <% } else if (inSessions.isEmpty() && outSession != null) { %>
+    <% } else if (inSessions.isEmpty() && !outSessions.isEmpty()) { %>
         <td width="1%">
             <img src="images/outgoing_32x16.gif" width="32" height="16" border="0" title="<fmt:message key='server.session.connection.outgoing' />" alt="<fmt:message key='server.session.connection.outgoing' />">
         </td>
@@ -68,7 +73,6 @@
 
     <% Date creationDate = null;
         Date lastActiveDate = null;
-        if (!inSessions.isEmpty() && outSession == null) {
         for (IncomingServerSession inSession : inSessions) {
             if (creationDate == null || creationDate.after(inSession.getCreationDate())) {
                 // Use the creation date of the oldest incoming session
@@ -79,24 +83,15 @@
                 lastActiveDate = inSession.getLastActiveDate();
             }
         }
-        }
-        else if (inSessions.isEmpty() && outSession != null) {
+        for (OutgoingServerSession outSession : outSessions) {
+            if (creationDate == null || creationDate.after(outSession.getCreationDate())) {
+                // Use the creation date of the oldest outgoing session
                 creationDate = outSession.getCreationDate();
-            lastActiveDate = outSession.getLastActiveDate();
-        }
-        else {
-            for (IncomingServerSession inSession : inSessions) {
-                if (creationDate == null || creationDate.after(inSession.getCreationDate())) {
-                    // Use the creation date of the oldest incoming session
-                    creationDate = inSession.getCreationDate();
-                }
-                if (lastActiveDate == null || lastActiveDate.before(inSession.getLastActiveDate())) {
-                    // Use the last active date of the newest incoming session
-                    lastActiveDate = inSession.getLastActiveDate();
             }
+            if (lastActiveDate == null || lastActiveDate.before(outSession.getLastActiveDate())) {
+                // Use the last active date of the newest outgoing session
+                lastActiveDate = outSession.getLastActiveDate();
             }
-            creationDate = creationDate.before(outSession.getCreationDate()) ? creationDate : outSession.getCreationDate();
-            lastActiveDate = lastActiveDate.after(outSession.getLastActiveDate()) ? lastActiveDate : outSession.getLastActiveDate();
         }
         Calendar creationCal = Calendar.getInstance();
         creationCal.setTime(creationDate);

src/web/server-session-summary.jsp

@@ -77,12 +77,14 @@
                 sess.close();
             }
 
-            Session sess = sessionManager.getOutgoingServerSession(hostname);
+            Collection<OutgoingServerSession> sessions = sessionManager.getOutgoingServerSessions(hostname);
+            for (OutgoingServerSession sess : sessions) {
                 if (sess != null) {
                     sess.close();
                 }
+            }
             // Log the event
-            webManager.logEvent("closed server session for "+hostname, null);
+            webManager.logEvent("closed server sessions for "+hostname, null);
             // wait one second
             Thread.sleep(1000L);
         }
@@ -194,8 +196,8 @@
         for (String host : hostnames) {
             count++;
             List<IncomingServerSession> inSessions = sessionManager.getIncomingServerSessions(host);
-            OutgoingServerSession outSession = sessionManager.getOutgoingServerSession(host);
-            if (inSessions.isEmpty() && outSession == null) {
+            List<OutgoingServerSession> outSessions = sessionManager.getOutgoingServerSessions(host);
+            if (inSessions.isEmpty() && outSessions.isEmpty()) {
                 // If the connections were just closed then skip this host
                 continue;
             }