package com.juick.xmpp.s2s; import com.juick.xmpp.Stanza; import com.juick.xmpp.StanzaChild; import com.juick.xmpp.extensions.JuickMessage; import org.springframework.jdbc.core.JdbcTemplate; import javax.servlet.ServletContextEvent; import javax.servlet.ServletContextListener; import java.io.IOException; import java.sql.Driver; import java.sql.DriverManager; import java.sql.SQLException; import java.util.*; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.logging.Level; import java.util.logging.Logger; /** * * @author ugnich */ public class XMPPComponent implements ServletContextListener { private static final Logger LOGGER = Logger.getLogger(XMPPComponent.class.getName()); public static final ExecutorService executorService = Executors.newWorkStealingPool(); public static String HOSTNAME = null; public static String STATSFILE = null; public static ConnectionRouter connRouter; static final List inConnections = Collections.synchronizedList(new ArrayList<>()); static final List outConnections = Collections.synchronizedList(new ArrayList<>()); static final List outCache = Collections.synchronizedList(new ArrayList<>()); static JdbcTemplate sql; final public static HashMap childParsers = new HashMap<>(); public static void addConnectionIn(ConnectionIn c) { synchronized (inConnections) { inConnections.add(c); } } public static void addConnectionOut(ConnectionOut c) { synchronized (outConnections) { outConnections.add(c); } } public static void removeConnectionIn(ConnectionIn c) { synchronized (inConnections) { inConnections.remove(c); } } public static void removeConnectionOut(ConnectionOut c) { synchronized (outConnections) { outConnections.remove(c); } } public static String getFromCache(String hostname) { CacheEntry ret = null; synchronized (outCache) { for (Iterator i = outCache.iterator(); i.hasNext();) { CacheEntry c = i.next(); if (c.hostname != null && c.hostname.equals(hostname)) { ret = c; i.remove(); break; } } } return (ret != null) ? ret.xml : null; } public static ConnectionOut getConnectionOut(String hostname, boolean needReady) { synchronized (outConnections) { for (ConnectionOut c : outConnections) { if (c.to != null && c.to.equals(hostname) && (!needReady || c.streamReady)) { return c; } } } return null; } public static ConnectionIn getConnectionIn(String streamID) { synchronized (inConnections) { for (ConnectionIn c : inConnections) { if (c.streamID != null && c.streamID.equals(streamID)) { return c; } } } return null; } public static void sendOut(Stanza s) { sendOut(s.to.Host, s.toString()); } public static void sendOut(String hostname, String xml) { boolean haveAnyConn = false; ConnectionOut connOut = null; synchronized (outConnections) { for (ConnectionOut c : outConnections) { if (c.to != null && c.to.equals(hostname)) { if (c.streamReady) { connOut = c; break; } else { haveAnyConn = true; break; } } } } if (connOut != null) { try { connOut.sendStanza(xml); } catch (IOException e) { LOGGER.warning("STREAM TO " + connOut.to + " " + connOut.streamID + " ERROR: " + e.toString()); } return; } boolean haveCache = false; synchronized (outCache) { for (CacheEntry c : outCache) { if (c.hostname != null && c.hostname.equals(hostname)) { c.xml += xml; c.tsUpdated = System.currentTimeMillis(); haveCache = true; break; } } if (!haveCache) { outCache.add(new CacheEntry(hostname, xml)); } } if (!haveAnyConn) { ConnectionOut connectionOut = new ConnectionOut(hostname); connectionOut.parseStream(); } } @Override public void contextInitialized(ServletContextEvent sce) { LOGGER.info("component initialized"); executorService.submit(() -> { Properties conf = new Properties(); try { conf.load(sce.getServletContext().getResourceAsStream("WEB-INF/s2s.conf")); HOSTNAME = conf.getProperty("hostname"); String componentName = conf.getProperty("componentname"); STATSFILE = conf.getProperty("statsfile"); sql = (JdbcTemplate) sce.getServletContext().getAttribute("sql"); childParsers.put(JuickMessage.XMLNS, new JuickMessage()); connRouter = new ConnectionRouter(componentName); executorService.submit(connRouter); executorService.submit(new ConnectionListener()); executorService.submit(new CleaningUp()); } catch (IOException e) { LOGGER.log(Level.SEVERE, "XMPPComponent error", e); } }); } @Override public void contextDestroyed(ServletContextEvent sce) { synchronized (XMPPComponent.outConnections) { for (Iterator i = XMPPComponent.outConnections.iterator(); i.hasNext();) { ConnectionOut c = i.next(); c.closeConnection(); i.remove(); } } synchronized (XMPPComponent.inConnections) { for (Iterator i = XMPPComponent.inConnections.iterator(); i.hasNext();) { ConnectionIn c = i.next(); c.closeConnection(); i.remove(); } } XMPPComponent.connRouter.closeConnection(); executorService.shutdown(); LOGGER.info("component destroyed"); } }