single xmpp component, WIP

3 files changed, 135 insertions, 67 deletions

diff --git a/src/main/java/com/juick/http/www/Main.java b/src/main/java/com/juick/http/www/Main.java
index 85abed00..36376d3c 100644
--- a/src/main/java/com/juick/http/www/Main.java
+++ b/src/main/java/com/juick/http/www/Main.java
@@ -17,10 +17,15 @@
  */
 package com.juick.http.www;
 
+import com.juick.CrosspostComponent;
+import com.juick.JuickComponent;
+import com.juick.JuickNotificator;
+import com.juick.PushComponent;
 import com.juick.server.UserQueries;
 import com.juick.xmpp.JID;
 import com.juick.xmpp.Stream;
 import com.juick.xmpp.StreamComponent;
+import com.juick.xmpp.s2s.S2SComponent;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.jdbc.datasource.DriverManagerDataSource;
 import ru.sape.Sape;
@@ -32,12 +37,20 @@ import javax.servlet.http.HttpServlet;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import java.io.IOException;
-import java.net.Socket;
+import java.net.InetSocketAddress;
 import java.net.URLEncoder;
-import java.sql.Connection;
+import java.nio.channels.AsynchronousSocketChannel;
+import java.nio.channels.Channels;
+import java.nio.channels.CompletionHandler;
+import java.sql.Driver;
 import java.sql.DriverManager;
 import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Enumeration;
+import java.util.List;
 import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 /**
  *
@@ -45,8 +58,8 @@ import java.util.Properties;
  */
 @WebServlet(name = "Main", urlPatterns = {"/"})
 @MultipartConfig(fileSizeThreshold = 1024 * 1024, maxRequestSize = 1024 * 1024 * 10)
-public class Main extends HttpServlet implements Stream.StreamListener {
-
+public class Main extends HttpServlet implements JuickNotificator, Stream.StreamListener {
+    static ExecutorService executorService;
     JdbcTemplate sql;
     JdbcTemplate sqlSearch;
     Stream xmpp;
@@ -64,6 +77,7 @@ public class Main extends HttpServlet implements Stream.StreamListener {
     SignUp signup = new SignUp();
     Settings settings = new Settings();
     RSS rss = new RSS();
+    static List<JuickComponent> components = new ArrayList<>();
 
     @Override
     public void init() throws ServletException {
@@ -73,6 +87,9 @@ public class Main extends HttpServlet implements Stream.StreamListener {
             Properties conf = new Properties();
             conf.load(getServletContext().getResourceAsStream("WEB-INF/juick.conf"));
 
+            executorService = Executors.newWorkStealingPool();
+            getServletContext().setAttribute("es", executorService);
+
             DriverManagerDataSource dataSource = new DriverManagerDataSource();
             dataSource.setDriverClassName(conf.getProperty("datasource_driver", "com.mysql.jdbc.Driver"));
             dataSource.setUrl(conf.getProperty("datasource_url"));
@@ -87,24 +104,43 @@ public class Main extends HttpServlet implements Stream.StreamListener {
             twitterAuth = new TwitterAuth(conf.getProperty("twitter_consumer_key"),
                     conf.getProperty("twitter_consumer_secret"));
             PageTemplates.sape = new Sape(conf.getProperty("sape_user"), "juick.com", 2000, 3600);
+            components.add(new S2SComponent(sql, executorService, conf));
+            components.add(new CrosspostComponent(sql, conf));
+            components.add(new PushComponent(sql, conf));
         } catch (Exception e) {
-            log(null, e);
+            log("www failed", e);
+        }
+    }
+
+    @Override
+    public void push(com.juick.xmpp.Message msg) {
+        for (JuickComponent c : components) {
+            c.messageReceived(msg);
         }
     }
 
     public void setupXmppComponent(final String password) {
-        Thread thr = new Thread(() -> {
+        executorService.submit(() -> {
             try {
-                Socket socket = new Socket("localhost", 5347);
-                xmpp = new StreamComponent(new JID("", "www.juick.com", ""), socket.getInputStream(),
-                        socket.getOutputStream(), password);
-                xmpp.addListener(Main.this);
-                xmpp.startParsing();
+                AsynchronousSocketChannel socket = AsynchronousSocketChannel.open();
+                socket.connect(new InetSocketAddress("localhost", 5347), socket, new CompletionHandler<Void, AsynchronousSocketChannel>() {
+                    @Override
+                    public void completed(Void result, AsynchronousSocketChannel attachment) {
+                        xmpp = new StreamComponent(new JID("", "www.juick.com", ""), Channels.newInputStream(socket),
+                                Channels.newOutputStream(socket), password);
+                        xmpp.addListener(Main.this);
+                        xmpp.startParsing();
+                    }
+
+                    @Override
+                    public void failed(Throwable exc, AsynchronousSocketChannel attachment) {
+                        log("www xmpp failed");
+                    }
+                });
             } catch (IOException e) {
                 log("xmpp exception", e);
             }
         });
-        thr.start();
     }
 
     @Override
@@ -311,4 +347,30 @@ public class Main extends HttpServlet implements Stream.StreamListener {
             response.sendError(405);
         }
     }
+
+    @Override
+    public void destroy() {
+        super.destroy();
+        executorService.shutdown();
+        // 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 {
+                    log(String.format("Deregistering JDBC driver %s", driver.toString()));
+                    DriverManager.deregisterDriver(driver);
+                } catch (SQLException ex) {
+                    log(String.format("Error deregistering JDBC driver %s", driver), ex);
+                }
+            } else {
+                // driver was not registered by the webapp's ClassLoader and may be in use elsewhere
+                log(String.format("Not deregistering JDBC driver %s as it does not belong to this webapp's ClassLoader", driver));
+            }
+        }
+    }
 }
diff --git a/src/main/java/com/juick/http/www/NewMessage.java b/src/main/java/com/juick/http/www/NewMessage.java
index 4d4ffde6..b3b63b17 100644
--- a/src/main/java/com/juick/http/www/NewMessage.java
+++ b/src/main/java/com/juick/http/www/NewMessage.java
@@ -31,6 +31,7 @@ import com.juick.xmpp.extensions.JuickUser;
 import com.juick.xmpp.extensions.Nickname;
 import com.juick.xmpp.extensions.XOOB;
 import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
 
 import java.io.IOException;
 import java.io.PrintWriter;
@@ -214,33 +215,32 @@ public class NewMessage {
             xoob.URL = attachmentURL;
             xmsg.addChild(xoob);
         }
-        if (xmpp != null) {
-            String tagsStr2 = "";
-            for (String tag : tagsArr) {
-                tagsStr2 += " *" + tag;
-            }
-            xmsg.body = "@" + jmsg.getUser().getUName() + ":" + tagsStr2 + "\n" + body + "\n\n#" + mid + " http://juick.com/" + mid;
-
-            xmsg.to = new JID("juick", "s2s.juick.com", null);
-            xmpp.send(xmsg);
+        String tagsStr2 = "";
+        for (String tag : tagsArr) {
+            tagsStr2 += " *" + tag;
+        }
+        xmsg.body = "@" + jmsg.getUser().getUName() + ":" + tagsStr2 + "\n" + body + "\n\n#" + mid + " http://juick.com/" + mid;
 
-            xmsg.to.Host = "ws.juick.com";
-            xmpp.send(xmsg);
+        xmsg.to = new JID("juick", "s2s.juick.com", null);
+        Main.push(xmsg);
 
-            xmsg.to.Host = "push.juick.com";
-            xmpp.send(xmsg);
+        xmsg.to.Host = "push.juick.com";
+        Main.push(xmsg);
 
-            xmsg.to.Host = "crosspost.juick.com";
-            xmsg.to.Username = "twitter";
-            xmpp.send(xmsg);
-            xmsg.to.Username = "fb";
-            xmpp.send(xmsg);
+        xmsg.to.Host = "crosspost.juick.com";
+        xmsg.to.Username = "twitter";
+        Main.push(xmsg);
+        xmsg.to.Username = "fb";
+        Main.push(xmsg);
 
-            xmsg.to.Host = "nologin.ru";
-            xmsg.to.Username = "jubo";
+        xmsg.to.Host = "nologin.ru";
+        xmsg.to.Username = "jubo";
+        Main.push(xmsg);
+        if (xmpp != null) {
+            xmsg.to.Host = "ws.juick.com";
             xmpp.send(xmsg);
         } else {
-            logger.warning("XMPP is not available, users will not be notified");
+            logger.warning("XMPP is not available, websockets will not be notified");
         }
 
         response.setContentType("text/html; charset=UTF-8");
@@ -375,19 +375,19 @@ public class NewMessage {
             xmsg.addChild(xoob);
         }
 
-        if (xmpp != null) {
-            xmsg.body = "Reply by @" + jmsg.getUser().getUName() + ":\n>" + quote + "\n" + body + "\n\n#" + mid + "/" + ridnew + " http://juick.com/" + mid + "#" + ridnew;
 
-            xmsg.to = new JID("juick", "s2s.juick.com", null);
-            xmpp.send(xmsg);
+        xmsg.body = "Reply by @" + jmsg.getUser().getUName() + ":\n>" + quote + "\n" + body + "\n\n#" + mid + "/" + ridnew + " http://juick.com/" + mid + "#" + ridnew;
 
-            xmsg.to.Host = "ws.juick.com";
-            xmpp.send(xmsg);
+        xmsg.to = new JID("juick", "s2s.juick.com", null);
+        Main.push(xmsg);
+        xmsg.to.Host = "push.juick.com";
+        Main.push(xmsg);
 
-            xmsg.to.Host = "push.juick.com";
+        if (xmpp != null) {
+            xmsg.to.Host = "ws.juick.com";
             xmpp.send(xmsg);
         } else {
-            logger.warning("XMPP is not available, users will not be notified");
+            logger.warning("XMPP is not available, websockets will not be notified");
         }
 
         Utils.sendTemporaryRedirect(response, "/" + msg.getUser().getUName() + "/" + mid + "#" + ridnew);
@@ -413,13 +413,12 @@ public class NewMessage {
 
         if (res) {
             Message xmsg = new Message();
-            xmsg.from = new JID("juick", "juick.com", null);
             xmsg.to = new JID("recomm", "s2s.juick.com", null);
             JuickMessage jmsg = new JuickMessage();
             jmsg.setMID(mid);
             jmsg.setUser(new JuickUser(visitor));
             xmsg.addChild(jmsg);
-            xmpp.send(xmsg);
+            Main.push(xmsg);
 
             Utils.replyJSON(request, response, "{\"status\":\"ok\"}");
         } else {
diff --git a/src/main/java/com/juick/http/www/PM.java b/src/main/java/com/juick/http/www/PM.java
index 1ee0b112..a6dfcec2 100644
--- a/src/main/java/com/juick/http/www/PM.java
+++ b/src/main/java/com/juick/http/www/PM.java
@@ -32,6 +32,7 @@ import java.io.IOException;
 import java.io.PrintWriter;
 import java.text.SimpleDateFormat;
 import java.util.List;
+import java.util.logging.Logger;
 
 /**
  *
@@ -39,6 +40,8 @@ import java.util.List;
  */
 public class PM {
 
+    private static final Logger logger = Logger.getLogger(PM.class.getName());
+
     private static final SimpleDateFormat sdfSQL = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
 
     protected void doGetInbox(JdbcTemplate sql, HttpServletRequest request, HttpServletResponse response, com.juick.User visitor) throws ServletException, IOException {
@@ -194,31 +197,35 @@ public class PM {
         }
 
         if (PMQueries.createPM(sql, visitor.getUID(), uid, body)) {
-            Message msg = new Message();
-            msg.from = new JID("juick", "juick.com", null);
-            msg.to = new JID(Integer.toString(uid), "push.juick.com", null);
-            JuickMessage jmsg = new JuickMessage();
-            jmsg.setUser(UserQueries.getUserByUID(sql, visitor.getUID()));
-            jmsg.setText(body);
-            msg.childs.add(jmsg);
-            xmpp.send(msg);
-
-            msg.to.Host = "ws.juick.com";
-            xmpp.send(msg);
-
-            String jid = UserQueries.getJIDbyUID(sql, uid);
-            if (jid != null) {
-                Message mm = new Message();
-                mm.to = new JID(jid);
-                mm.type = Message.Type.chat;
-                if (PMQueries.havePMinRoster(sql, visitor.getUID(), jid)) {
-                    mm.from = new JID(jmsg.getUser().getUName(), "juick.com", "Juick");
-                    mm.body = body;
-                } else {
-                    mm.from = new JID("juick", "juick.com", "Juick");
-                    mm.body = "Private message from @" + jmsg.getUser().getUName() + ":\n" + body;
+            if (xmpp != null) {
+                Message msg = new Message();
+                msg.from = new JID("juick", "juick.com", null);
+                msg.to = new JID(Integer.toString(uid), "push.juick.com", null);
+                JuickMessage jmsg = new JuickMessage();
+                jmsg.setUser(UserQueries.getUserByUID(sql, visitor.getUID()));
+                jmsg.setText(body);
+                msg.childs.add(jmsg);
+                xmpp.send(msg);
+
+                msg.to.Host = "ws.juick.com";
+                xmpp.send(msg);
+
+                String jid = UserQueries.getJIDbyUID(sql, uid);
+                if (jid != null) {
+                    Message mm = new Message();
+                    mm.to = new JID(jid);
+                    mm.type = Message.Type.chat;
+                    if (PMQueries.havePMinRoster(sql, visitor.getUID(), jid)) {
+                        mm.from = new JID(jmsg.getUser().getUName(), "juick.com", "Juick");
+                        mm.body = body;
+                    } else {
+                        mm.from = new JID("juick", "juick.com", "Juick");
+                        mm.body = "Private message from @" + jmsg.getUser().getUName() + ":\n" + body;
+                    }
+                    xmpp.send(mm);
                 }
-                xmpp.send(mm);
+            } else {
+                logger.warning("XMPP is not available");
             }
 
             Utils.sendTemporaryRedirect(response, "/pm/sent");