diff options
Diffstat (limited to 'src/main/java/com/juick/xmpp/s2s/XMPPComponent.java')
| -rw-r--r-- | src/main/java/com/juick/xmpp/s2s/XMPPComponent.java | 377 |
1 files changed, 284 insertions, 93 deletions
diff --git a/src/main/java/com/juick/xmpp/s2s/XMPPComponent.java b/src/main/java/com/juick/xmpp/s2s/XMPPComponent.java index 13df5fd0..d0b231e2 100644 --- a/src/main/java/com/juick/xmpp/s2s/XMPPComponent.java +++ b/src/main/java/com/juick/xmpp/s2s/XMPPComponent.java @@ -1,21 +1,27 @@ package com.juick.xmpp.s2s; -import com.juick.xmpp.Stanza; -import com.juick.xmpp.StanzaChild; +import com.juick.User; +import com.juick.server.MessagesQueries; +import com.juick.server.SubscriptionsQueries; +import com.juick.server.UserQueries; +import com.juick.xmpp.*; import com.juick.xmpp.extensions.JuickMessage; +import com.juick.xmpp.extensions.Nickname; +import com.juick.xmpp.extensions.XOOB; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DriverManagerDataSource; import org.xmlpull.v1.XmlPullParserException; import javax.servlet.ServletContextEvent; import javax.servlet.ServletContextListener; import java.io.IOException; +import java.net.ServerSocket; +import java.net.Socket; import java.security.KeyManagementException; import java.security.KeyStoreException; import java.security.NoSuchAlgorithmException; import java.security.UnrecoverableKeyException; import java.security.cert.CertificateException; -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; @@ -26,50 +32,51 @@ 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.newCachedThreadPool(); - - public static String HOSTNAME = null; - public static String STATSFILE = null; - public static String keystore; - public static String keystorePassword; - public static List<String> brokenSSLhosts; - public static ConnectionRouter connRouter; - static final List<ConnectionIn> inConnections = Collections.synchronizedList(new ArrayList<>()); - static final List<ConnectionOut> outConnections = Collections.synchronizedList(new ArrayList<>()); - static final List<CacheEntry> outCache = Collections.synchronizedList(new ArrayList<>()); - static final Integer sqlSync = 0; - static java.sql.Connection sql; - final public static HashMap<String, StanzaChild> childParsers = new HashMap<>(); - - public static void addConnectionIn(ConnectionIn c) { +public class XMPPComponent implements ServletContextListener, Stream.StreamListener, + Message.MessageListener, Iq.IqListener, Presence.PresenceListener { + + private static final Logger logger = Logger.getLogger(XMPPComponent.class.getName()); + + public final ExecutorService executorService = Executors.newCachedThreadPool(); + StreamComponent router; + JuickBot bot; + + public String HOSTNAME, componentName; + public String STATSFILE = null; + public String keystore; + public String keystorePassword; + public List<String> brokenSSLhosts; + final List<ConnectionIn> inConnections = Collections.synchronizedList(new ArrayList<>()); + final List<ConnectionOut> outConnections = Collections.synchronizedList(new ArrayList<>()); + final List<CacheEntry> outCache = Collections.synchronizedList(new ArrayList<>()); + JdbcTemplate sql; + final public HashMap<String, StanzaChild> childParsers = new HashMap<>(); + + public void addConnectionIn(ConnectionIn c) { synchronized (inConnections) { inConnections.add(c); } } - public static void addConnectionOut(ConnectionOut c) { + public void addConnectionOut(ConnectionOut c) { synchronized (outConnections) { outConnections.add(c); } } - public static void removeConnectionIn(ConnectionIn c) { + public void removeConnectionIn(ConnectionIn c) { synchronized (inConnections) { inConnections.remove(c); } } - public static void removeConnectionOut(ConnectionOut c) { + public void removeConnectionOut(ConnectionOut c) { synchronized (outConnections) { outConnections.remove(c); } } - public static String getFromCache(String hostname) { + public String getFromCache(String hostname) { CacheEntry ret = null; synchronized (outCache) { for (Iterator<CacheEntry> i = outCache.iterator(); i.hasNext();) { @@ -84,7 +91,7 @@ public class XMPPComponent implements ServletContextListener { return (ret != null) ? ret.xml : null; } - public static ConnectionOut getConnectionOut(String hostname, boolean needReady) { + public ConnectionOut getConnectionOut(String hostname, boolean needReady) { synchronized (outConnections) { for (ConnectionOut c : outConnections) { if (c.to != null && c.to.equals(hostname) && (!needReady || c.streamReady)) { @@ -95,7 +102,7 @@ public class XMPPComponent implements ServletContextListener { return null; } - public static ConnectionIn getConnectionIn(String streamID) { + public ConnectionIn getConnectionIn(String streamID) { synchronized (inConnections) { for (ConnectionIn c : inConnections) { if (c.streamID != null && c.streamID.equals(streamID)) { @@ -106,11 +113,11 @@ public class XMPPComponent implements ServletContextListener { return null; } - public static void sendOut(Stanza s) { + public void sendOut(Stanza s) { sendOut(s.to.Host, s.toString()); } - public static void sendOut(String hostname, String xml) { + public void sendOut(String hostname, String xml) { boolean haveAnyConn = false; ConnectionOut connOut = null; @@ -131,7 +138,7 @@ public class XMPPComponent implements ServletContextListener { try { connOut.sendStanza(xml); } catch (IOException e) { - LOGGER.warning("STREAM TO " + connOut.to + " " + connOut.streamID + " ERROR: " + e.toString()); + logger.warning("STREAM TO " + connOut.to + " " + connOut.streamID + " ERROR: " + e.toString()); } return; } @@ -152,12 +159,11 @@ public class XMPPComponent implements ServletContextListener { } if (!haveAnyConn) { - ConnectionOut connectionOut = null; try { - connectionOut = new ConnectionOut(hostname); - XMPPComponent.executorService.submit(connectionOut); + ConnectionOut connectionOut = new ConnectionOut(this, hostname); + executorService.submit(connectionOut); } catch (CertificateException | UnrecoverableKeyException | NoSuchAlgorithmException | XmlPullParserException | KeyStoreException | KeyManagementException | IOException e) { - LOGGER.log(Level.SEVERE, "s2s out error", e); + logger.log(Level.SEVERE, "s2s out error", e); } } } @@ -165,45 +171,74 @@ public class XMPPComponent implements ServletContextListener { @Override public void contextInitialized(ServletContextEvent sce) { - LOGGER.info("component initialized"); - executorService.submit(() -> { - Properties conf = new Properties(); - try { - conf.load(sce.getServletContext().getResourceAsStream("/WEB-INF/juick.conf")); - HOSTNAME = conf.getProperty("hostname"); - String componentName = conf.getProperty("componentname"); - STATSFILE = conf.getProperty("statsfile"); - keystore = conf.getProperty("keystore"); - keystorePassword = conf.getProperty("keystore_password"); - brokenSSLhosts = Arrays.asList(conf.getProperty("broken_ssl_hosts", "").split(",")); - Class.forName("com.mysql.jdbc.Driver"); - sql = DriverManager.getConnection("jdbc:mysql://localhost/juick?autoReconnect=true&user=" + - conf.getProperty("mysql_username", "") + "&password=" + conf.getProperty("mysql_password", "")); - - childParsers.put(JuickMessage.XMLNS, new JuickMessage()); - executorService.submit(() -> connRouter = new ConnectionRouter(componentName, conf.getProperty("xmpp_password"))); - executorService.submit(new ConnectionListener()); - executorService.submit(new CleaningUp()); - } catch (Exception e) { - LOGGER.log(Level.SEVERE, "XMPPComponent error", e); - } - }); + logger.info("component initialized"); + Properties conf = new Properties(); + try { + conf.load(sce.getServletContext().getResourceAsStream("/WEB-INF/juick.conf")); + HOSTNAME = conf.getProperty("hostname"); + componentName = conf.getProperty("componentname"); + JID Jid = new JID(conf.getProperty("xmppbot_jid")); + STATSFILE = conf.getProperty("statsfile"); + keystore = conf.getProperty("keystore"); + keystorePassword = conf.getProperty("keystore_password"); + brokenSSLhosts = Arrays.asList(conf.getProperty("broken_ssl_hosts", "").split(",")); + DriverManagerDataSource dataSource = new DriverManagerDataSource(); + dataSource.setDriverClassName(conf.getProperty("datasource_driver", "com.mysql.jdbc.Driver")); + dataSource.setUrl(conf.getProperty("datasource_url")); + sql = new JdbcTemplate(dataSource); + bot = new JuickBot(this, Jid); + + childParsers.put(JuickMessage.XMLNS, new JuickMessage()); + + executorService.submit(() -> { + Socket routerSocket = null; + try { + routerSocket = new Socket("localhost", 5347); + router = new StreamComponent(new JID("s2s"), routerSocket.getInputStream(), routerSocket.getOutputStream(), conf.getProperty("xmpp_password")); + router.addChildParser(new JuickMessage()); + router.addListener((Stream.StreamListener) this); + router.addListener((Message.MessageListener) this); + router.addListener((Iq.IqListener) this); + router.startParsing(); + } catch (IOException e) { + logger.log(Level.SEVERE, "router error", e); + } + }); + executorService.submit(() -> { + final ServerSocket listener = new ServerSocket(5269); + logger.info("s2s listener ready"); + while (true) { + try { + Socket socket = listener.accept(); + ConnectionIn client = new ConnectionIn(this, bot, socket); + addConnectionIn(client); + executorService.submit(client); + } catch (Exception e) { + logger.log(Level.SEVERE, "s2s error", e); + } + } + }); + executorService.submit(new CleaningUp(this)); + + } catch (Exception e) { + logger.log(Level.SEVERE, "XMPPComponent error", e); + } } @Override public void contextDestroyed(ServletContextEvent sce) { - synchronized (XMPPComponent.outConnections) { - for (Iterator<ConnectionOut> i = XMPPComponent.outConnections.iterator(); i.hasNext();) { + synchronized (outConnections) { + for (Iterator<ConnectionOut> i = outConnections.iterator(); i.hasNext();) { ConnectionOut c = i.next(); c.closeConnection(); i.remove(); } } - synchronized (XMPPComponent.inConnections) { - for (Iterator<ConnectionIn> i = XMPPComponent.inConnections.iterator(); i.hasNext();) { + synchronized (inConnections) { + for (Iterator<ConnectionIn> i = inConnections.iterator(); i.hasNext();) { ConnectionIn c = i.next(); c.closeConnection(); i.remove(); @@ -211,42 +246,198 @@ public class XMPPComponent implements ServletContextListener { } try { - XMPPComponent.connRouter.closeConnection(); + closeRouterConnection(); } catch (IOException e) { - LOGGER.log(Level.WARNING, "router warning", e); + logger.log(Level.WARNING, "router warning", e); } + executorService.shutdown(); + logger.info("component destroyed"); + } + public void closeRouterConnection() throws IOException { + router.logoff(); + } - synchronized (XMPPComponent.sqlSync) { - if (XMPPComponent.sql != null) { - try { - XMPPComponent.sql.close(); - XMPPComponent.sql = null; - } catch (SQLException e) { - LOGGER.warning("SQL ERROR: " + e); + public void sendJuickMessage(JuickMessage jmsg) { + List<String> jids = new ArrayList<>(); + + if (jmsg.FriendsOnly) { + jids = SubscriptionsQueries.getJIDSubscribedToUser(sql, jmsg.getUser().getUID(), jmsg.FriendsOnly); + } else { + List<User> users = SubscriptionsQueries.getSubscribedUsers(sql, jmsg.getUser().getUID(), jmsg.getMID()); + for (User user : users) { + for (String jid : UserQueries.getJIDsbyUID(sql, user.getUID())) { + jids.add(jid); } } } - // Now deregister JDBC drivers in this context's ClassLoader: - // Get the webapp's ClassLoader - ClassLoader cl = Thread.currentThread().getContextClassLoader(); - // Loop through all drivers - Enumeration<Driver> drivers = DriverManager.getDrivers(); - while (drivers.hasMoreElements()) { - Driver driver = drivers.nextElement(); - if (driver.getClass().getClassLoader() == cl) { - // This driver was registered by the webapp's ClassLoader, so deregister it: - try { - LOGGER.info(String.format("Deregistering JDBC driver %s", driver.toString())); - DriverManager.deregisterDriver(driver); - } catch (SQLException ex) { - LOGGER.log(Level.SEVERE, String.format("Error deregistering JDBC driver %s", driver), ex); - } + + String txt = "@" + jmsg.getUser().getUName() + ":" + jmsg.getTagsString() + "\n"; + String attachment = jmsg.getAttachmentURL(); + if (attachment != null) { + txt += attachment + "\n"; + } + txt += jmsg.getText() + "\n\n"; + txt += "#" + jmsg.getMID() + " http://juick.com/" + jmsg.getMID(); + + Nickname nick = new Nickname(); + nick.Nickname = "@" + jmsg.getUser().getUName(); + + com.juick.xmpp.Message msg = new com.juick.xmpp.Message(); + msg.from = bot.JuickJID; + msg.body = txt; + msg.type = Message.Type.chat; + msg.thread = "juick-" + jmsg.getMID(); + msg.addChild(jmsg); + msg.addChild(nick); + if (attachment != null) { + XOOB oob = new XOOB(); + oob.URL = attachment; + msg.addChild(oob); + } + + for (String jid : jids) { + msg.to = new JID(jid); + sendOut(msg); + } + } + + public void sendJuickComment(JuickMessage jmsg) { + List<User> users; + String replyQuote; + String replyTo; + + users = SubscriptionsQueries.getUsersSubscribedToComments(sql, jmsg.getMID(), jmsg.getUser().getUID()); + com.juick.Message replyMessage = jmsg.ReplyTo > 0 ? MessagesQueries.getReply(sql, jmsg.getMID(), jmsg.ReplyTo) + : MessagesQueries.getMessage(sql, jmsg.getMID()); + replyTo = replyMessage.getUser().getUName(); + replyQuote = getReplyQuote(replyMessage); + + String txt = "Reply by @" + jmsg.getUser().getUName() + ":\n" + replyQuote + "\n@" + replyTo + " "; + String attachment = jmsg.getAttachmentURL(); + if (attachment != null) { + txt += attachment + "\n"; + } + txt += jmsg.getText() + "\n\n" + "#" + jmsg.getMID() + "/" + jmsg.getRID() + " http://juick.com/" + jmsg.getMID() + "#" + jmsg.getRID(); + + com.juick.xmpp.Message msg = new com.juick.xmpp.Message(); + msg.from = bot.JuickJID; + msg.body = txt; + msg.type = Message.Type.chat; + msg.addChild(jmsg); + for (User user : users) { + for (String jid : UserQueries.getJIDsbyUID(sql, user.getUID())) { + msg.to = new JID(jid); + sendOut(msg); + } + } + } + + private String getReplyQuote(com.juick.Message q) { + String quote = q.getText(); + if (quote.length() > 50) { + quote = ">" + quote.substring(0, 47).replace('\n', ' ') + "...\n"; + } else if (quote.length() > 0) { + quote = ">" + quote.replace('\n', ' ') + "\n"; + } + return quote; + } + + public void sendJuickRecommendation(JuickMessage recomm) { + List<User> users; + JuickMessage jmsg; + jmsg = new JuickMessage(MessagesQueries.getMessage(sql, recomm.getMID())); + users = SubscriptionsQueries.getUsersSubscribedToUserRecommendations(sql, + recomm.getUser().getUID(), recomm.getMID(), jmsg.getUser().getUID()); + + String txt = "Recommended by @" + recomm.getUser().getUName() + ":\n"; + txt += "@" + jmsg.getUser().getUName() + ":" + jmsg.getTagsString() + "\n"; + String attachment = jmsg.getAttachmentURL(); + if (attachment != null) { + txt += attachment + "\n"; + } + txt += jmsg.getText() + "\n\n"; + txt += "#" + jmsg.getMID(); + if (jmsg.Replies > 0) { + if (jmsg.Replies % 10 == 1 && jmsg.Replies % 100 != 11) { + txt += " (" + jmsg.Replies + " reply)"; } else { - // driver was not registered by the webapp's ClassLoader and may be in use elsewhere - LOGGER.log(Level.SEVERE, String.format("Not deregistering JDBC driver %s as it does not belong to this webapp's ClassLoader", driver)); + txt += " (" + jmsg.Replies + " replies)"; } } - executorService.shutdown(); - LOGGER.info("component destroyed"); + txt += " http://juick.com/" + jmsg.getMID(); + + Nickname nick = new Nickname(); + nick.Nickname = "@" + jmsg.getUser().getUName(); + + com.juick.xmpp.Message msg = new com.juick.xmpp.Message(); + msg.from = bot.JuickJID; + msg.body = txt; + msg.type = Message.Type.chat; + msg.thread = "juick-" + jmsg.getMID(); + msg.addChild(jmsg); + msg.addChild(nick); + if (attachment != null) { + XOOB oob = new XOOB(); + oob.URL = attachment; + msg.addChild(oob); + } + + for (User user : users) { + for (String jid : UserQueries.getJIDsbyUID(sql, user.getUID())) { + msg.to = new JID(jid); + sendOut(msg); + } + } + } + + @Override + public boolean onIq(Iq iq) { + JID jid = iq.to; + if (!jid.Host.equals(componentName)) { + logger.info("STREAM ROUTER (IQ): " + iq.toString()); + sendOut(iq); + } + return false; + } + + @Override + public void onMessage(Message xmsg) { + logger.info("STREAM ROUTER (PROCESS): " + xmsg.toString()); + JuickMessage jmsg = (JuickMessage) xmsg.getChild(JuickMessage.XMLNS); + JID jid = xmsg.to; + if (jid.Host.equals(componentName)) { + if (jmsg != null) { + if (jid.Username != null && jid.Username.equals("recomm")) { + sendJuickRecommendation(jmsg); + } else { + if (jmsg.getRID() > 0) { + sendJuickComment(jmsg); + } else if (jmsg.getMID() > 0) { + sendJuickMessage(jmsg); + } + } + } + } else { + sendOut(xmsg); + } + } + + @Override + public void onPresence(Presence presence) { + JID jid = presence.to; + if (!jid.Host.equals(componentName)) { + logger.info("STREAM ROUTER (PRESENCE): " + presence.toString()); + sendOut(presence); + } + } + + @Override + public void onStreamReady() { + logger.info("STREAM ROUTER (READY)"); + } + + @Override + public void onStreamFail(Exception ex) { + logger.log(Level.SEVERE, "STREAM ROUTER (FAIL)", ex); } } |