aboutsummaryrefslogtreecommitdiff
path: root/src/main/java/com/juick/xmpp/s2s
diff options
context:
space:
mode:
authorGravatar Vitaly Takmazov2016-06-07 09:39:35 +0300
committerGravatar Vitaly Takmazov2016-06-07 09:39:35 +0300
commit7db1637114eaaf36fb229999b478abc0bb3b1ab5 (patch)
tree995f7e591e60f97697983f2e3832c25ee0a4b718 /src/main/java/com/juick/xmpp/s2s
parente7e9bed758028e1555c9cbbcaaa29fe5504db280 (diff)
refactoring
Diffstat (limited to 'src/main/java/com/juick/xmpp/s2s')
-rw-r--r--src/main/java/com/juick/xmpp/s2s/Connection.java6
-rw-r--r--src/main/java/com/juick/xmpp/s2s/ConnectionIn.java4
-rw-r--r--src/main/java/com/juick/xmpp/s2s/ConnectionOut.java8
-rw-r--r--src/main/java/com/juick/xmpp/s2s/ConnectionRouter.java189
-rw-r--r--src/main/java/com/juick/xmpp/s2s/JuickBot.java8
-rw-r--r--src/main/java/com/juick/xmpp/s2s/XMPPComponent.java29
6 files changed, 118 insertions, 126 deletions
diff --git a/src/main/java/com/juick/xmpp/s2s/Connection.java b/src/main/java/com/juick/xmpp/s2s/Connection.java
index c3e983b5..c1bbf22e 100644
--- a/src/main/java/com/juick/xmpp/s2s/Connection.java
+++ b/src/main/java/com/juick/xmpp/s2s/Connection.java
@@ -12,8 +12,8 @@ import javax.net.ssl.TrustManager;
import javax.net.ssl.X509TrustManager;
import java.io.*;
import java.net.Socket;
-import java.security.KeyStore;
-import java.security.SecureRandom;
+import java.security.*;
+import java.security.cert.CertificateException;
import java.util.UUID;
import java.util.logging.Logger;
@@ -52,7 +52,7 @@ public class Connection {
};
- public Connection() throws Exception {
+ public Connection() throws XmlPullParserException, KeyStoreException, CertificateException, NoSuchAlgorithmException, IOException, UnrecoverableKeyException, KeyManagementException {
tsCreated = System.currentTimeMillis();
parser.setFeature(XmlPullParser.FEATURE_PROCESS_NAMESPACES, true);
KeyStore ks = KeyStore.getInstance("JKS");
diff --git a/src/main/java/com/juick/xmpp/s2s/ConnectionIn.java b/src/main/java/com/juick/xmpp/s2s/ConnectionIn.java
index 554d3b05..93e761a0 100644
--- a/src/main/java/com/juick/xmpp/s2s/ConnectionIn.java
+++ b/src/main/java/com/juick/xmpp/s2s/ConnectionIn.java
@@ -115,7 +115,7 @@ public class ConnectionIn extends Connection implements Runnable {
if (msg != null && (msg.type == null || !msg.type.equals(Message.Type.error))) {
LOGGER.info("STREAM " + streamID + ": " + msg.toString());
if (!JuickBot.incomingMessage(msg)) {
- XMPPComponent.connRouter.sendStanza(msg.toString());
+ XMPPComponent.connRouter.router.send(msg.toString());
}
}
} else if (tag.equals("iq") && checkFromTo(parser)) {
@@ -124,7 +124,7 @@ public class ConnectionIn extends Connection implements Runnable {
String xml = XmlUtils.parseToString(parser, true);
if (type == null || !type.equals(Iq.Type.error)) {
LOGGER.info("STREAM " + streamID + ": " + xml);
- XMPPComponent.connRouter.sendStanza(xml);
+ XMPPComponent.connRouter.router.send(xml);
}
} else if (!isSecured() && tag.equals("starttls")) {
LOGGER.info("STREAM " + streamID + " SECURING");
diff --git a/src/main/java/com/juick/xmpp/s2s/ConnectionOut.java b/src/main/java/com/juick/xmpp/s2s/ConnectionOut.java
index 68851da1..d3a10406 100644
--- a/src/main/java/com/juick/xmpp/s2s/ConnectionOut.java
+++ b/src/main/java/com/juick/xmpp/s2s/ConnectionOut.java
@@ -3,6 +3,7 @@ package com.juick.xmpp.s2s;
import com.juick.xmpp.extensions.StreamFeatures;
import com.juick.xmpp.utils.XmlUtils;
import org.xmlpull.v1.XmlPullParser;
+import org.xmlpull.v1.XmlPullParserException;
import javax.net.ssl.SSLException;
import javax.net.ssl.SSLSocket;
@@ -11,6 +12,11 @@ import java.io.IOException;
import java.net.InetAddress;
import java.net.Socket;
import java.net.UnknownHostException;
+import java.security.KeyManagementException;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.UnrecoverableKeyException;
+import java.security.cert.CertificateException;
import java.util.logging.Level;
/**
@@ -23,7 +29,7 @@ public class ConnectionOut extends Connection implements Runnable {
String checkSID = null;
String dbKey = null;
- public ConnectionOut(String hostname) throws Exception {
+ public ConnectionOut(String hostname) throws CertificateException, UnrecoverableKeyException, NoSuchAlgorithmException, XmlPullParserException, KeyManagementException, KeyStoreException, IOException {
super();
to = hostname;
}
diff --git a/src/main/java/com/juick/xmpp/s2s/ConnectionRouter.java b/src/main/java/com/juick/xmpp/s2s/ConnectionRouter.java
index 61ae89e0..443bfa82 100644
--- a/src/main/java/com/juick/xmpp/s2s/ConnectionRouter.java
+++ b/src/main/java/com/juick/xmpp/s2s/ConnectionRouter.java
@@ -2,127 +2,49 @@ package com.juick.xmpp.s2s;
import com.juick.server.MessagesQueries;
import com.juick.server.SubscriptionsQueries;
-import com.juick.xmpp.JID;
-import com.juick.xmpp.Message;
+import com.juick.xmpp.*;
import com.juick.xmpp.extensions.JuickMessage;
import com.juick.xmpp.extensions.Nickname;
import com.juick.xmpp.extensions.XOOB;
-import com.juick.xmpp.utils.XmlUtils;
-import org.apache.commons.codec.digest.DigestUtils;
-import org.xmlpull.v1.XmlPullParser;
import java.io.IOException;
-import java.io.InputStreamReader;
-import java.io.OutputStreamWriter;
import java.net.Socket;
import java.util.List;
import java.util.logging.Level;
+import java.util.logging.Logger;
/**
* @author ugnich
*/
-public class ConnectionRouter extends Connection implements Runnable {
+public class ConnectionRouter implements Stream.StreamListener,
+ Message.MessageListener, Iq.IqListener, Presence.PresenceListener {
+ private static final Logger logger = Logger.getLogger(ConnectionRouter.class.getName());
private String componentName;
+ Stream router;
+ Socket socket;
- ConnectionRouter(String componentName) throws Exception {
+ ConnectionRouter(String componentName, String password) {
this.componentName = componentName;
- }
-
- @Override
- public void run() {
- LOGGER.info("STREAM ROUTER START");
-
+ logger.info("STREAM ROUTER START");
try {
socket = new Socket("localhost", 5347);
- parser.setInput(new InputStreamReader(socket.getInputStream()));
-
- parser.setFeature(XmlPullParser.FEATURE_PROCESS_NAMESPACES, true);
- writer = new OutputStreamWriter(socket.getOutputStream());
-
- String msg = "<stream:stream xmlns='jabber:component:accept' xmlns:stream='http://etherx.jabber.org/streams' to='s2s'>";
- writer.write(msg);
- writer.flush();
-
- parser.next(); // stream:stream
- streamID = parser.getAttributeValue(null, "id");
- if (streamID == null || streamID.isEmpty()) {
- throw new Exception("FAIL ON FIRST PACKET");
- }
-
- msg = "<handshake>" + DigestUtils.sha1Hex(streamID + "secret") + "</handshake>";
- writer.write(msg);
- writer.flush();
-
- parser.next();
- if (!parser.getName().equals("handshake")) {
- throw new Exception("NO HANDSHAKE");
- }
- XmlUtils.skip(parser);
- LOGGER.info("STREAM ROUTER OPEN");
-
- while (parser.next() != XmlPullParser.END_DOCUMENT) {
- if (parser.getEventType() != XmlPullParser.START_TAG) {
- continue;
- }
-
- String tag = parser.getName();
- String to = parser.getAttributeValue(null, "to");
- if (to != null && (tag.equals("message") || tag.equals("presence") || tag.equals("iq"))) {
- JID jid = new JID(to);
- if (jid.Host != null) {
- if (jid.Host.equals(componentName)) {
- if (tag.equals("message")) {
- Message xmsg = Message.parse(parser, XMPPComponent.childParsers);
- LOGGER.info("STREAM ROUTER (PROCESS): " + xmsg.toString());
- JuickMessage jmsg = (JuickMessage) xmsg.getChild(JuickMessage.XMLNS);
- 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 if (jid.Host.endsWith(XMPPComponent.HOSTNAME) && (jid.Host.equals(XMPPComponent.HOSTNAME) || jid.Host.endsWith("." + XMPPComponent.HOSTNAME))) {
- String xml = XmlUtils.parseToString(parser, true);
- LOGGER.info("STREAM ROUTER: " + xml);
- } else {
- String xml = XmlUtils.parseToString(parser, true);
- LOGGER.info("STREAM ROUTER (OUT): " + xml);
- XMPPComponent.sendOut(jid.Host, xml);
- }
- } else {
- LOGGER.info("STREAM ROUTER (NO TO): " + XmlUtils.parseToString(parser, true));
- }
- } else {
- LOGGER.info("STREAM ROUTER: " + XmlUtils.parseToString(parser, true));
- }
- }
-
- LOGGER.warning("STREAM ROUTER FINISHED");
- } catch (Exception e) {
- LOGGER.log(Level.SEVERE, "xmpp router exception", e);
- }
- }
-
- @Override
- synchronized public void sendStanza(String xml) {
- try {
- writer.write(xml);
- writer.flush();
+ router = new StreamComponent(new JID(componentName), socket.getInputStream(), socket.getOutputStream(), 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.warning("STREAM ROUTER ERROR: " + xml);
- LOGGER.warning("STREAM ROUTER ERROR: " + e.toString());
- System.exit(0);
+ logger.log(Level.SEVERE, "router failed", e);
}
}
+ public void closeConnection() throws IOException {
+ router.logoff();
+ socket.close();
+ }
- public void sendJuickMessage(JuickMessage jmsg) throws Exception {
+ public void sendJuickMessage(JuickMessage jmsg) {
List<String> jids;
synchronized (XMPPComponent.sqlSync) {
@@ -157,16 +79,16 @@ public class ConnectionRouter extends Connection implements Runnable {
msg.addChild(oob);
}
- for (int i = 0; i < jids.size(); i++) {
- msg.to = new JID(jids.get(i));
+ for (String jid : jids) {
+ msg.to = new JID(jid);
XMPPComponent.sendOut(msg);
}
}
- public void sendJuickComment(JuickMessage jmsg) throws Exception {
+ public void sendJuickComment(JuickMessage jmsg) {
List<String> jids;
String replyQuote;
- String replyTo;
+ String replyTo;
synchronized (XMPPComponent.sqlSync) {
jids = SubscriptionsQueries.getJIDSubscribedToComments(XMPPComponent.sql, jmsg.getMID(), jmsg.getUser().getUID());
@@ -188,8 +110,8 @@ public class ConnectionRouter extends Connection implements Runnable {
msg.body = txt;
msg.type = Message.Type.chat;
msg.addChild(jmsg);
- for (int i = 0; i < jids.size(); i++) {
- msg.to = new JID(jids.get(i));
+ for (String jid : jids) {
+ msg.to = new JID(jid);
XMPPComponent.sendOut(msg);
}
}
@@ -204,7 +126,7 @@ public class ConnectionRouter extends Connection implements Runnable {
return quote;
}
- public void sendJuickRecommendation(JuickMessage recomm) throws Exception {
+ public void sendJuickRecommendation(JuickMessage recomm) {
List<String> jids;
JuickMessage jmsg;
synchronized (XMPPComponent.sqlSync) {
@@ -246,9 +168,60 @@ public class ConnectionRouter extends Connection implements Runnable {
msg.addChild(oob);
}
- for (int i = 0; i < jids.size(); i++) {
- msg.to = new JID(jids.get(i));
+ for (String jid : jids) {
+ msg.to = new JID(jid);
XMPPComponent.sendOut(msg);
}
}
+
+ @Override
+ public boolean onIq(Iq iq) {
+ JID jid = iq.to;
+ if (!jid.Host.equals(componentName)) {
+ logger.info("STREAM ROUTER (IQ): " + iq.toString());
+ XMPPComponent.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 {
+ XMPPComponent.sendOut(xmsg);
+ }
+ }
+
+ @Override
+ public void onPresence(Presence presence) {
+ JID jid = presence.to;
+ if (!jid.Host.equals(componentName)) {
+ logger.info("STREAM ROUTER (PRESENCE): " + presence.toString());
+ XMPPComponent.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);
+ }
}
diff --git a/src/main/java/com/juick/xmpp/s2s/JuickBot.java b/src/main/java/com/juick/xmpp/s2s/JuickBot.java
index 0b9cceaf..25c75dfe 100644
--- a/src/main/java/com/juick/xmpp/s2s/JuickBot.java
+++ b/src/main/java/com/juick/xmpp/s2s/JuickBot.java
@@ -198,10 +198,10 @@ public class JuickBot {
}
jmsg.setText(msg.body);
m.childs.add(jmsg);
- XMPPComponent.connRouter.sendStanza(m.toString());
+ XMPPComponent.connRouter.router.send(m.toString());
m.to.Host = "ws.juick.com";
- XMPPComponent.connRouter.sendStanza(m.toString());
+ XMPPComponent.connRouter.router.send(m.toString());
String jid;
boolean inroster = false;
@@ -337,10 +337,10 @@ public class JuickBot {
jmsg.setUser(user_from);
jmsg.setText(body);
msg.childs.add(jmsg);
- XMPPComponent.connRouter.sendStanza(msg.toString());
+ XMPPComponent.connRouter.router.send(msg.toString());
msg.to.Host = "ws.juick.com";
- XMPPComponent.connRouter.sendStanza(msg.toString());
+ XMPPComponent.connRouter.router.send(msg.toString());
if (jid_to != null) {
Message mm = new Message();
diff --git a/src/main/java/com/juick/xmpp/s2s/XMPPComponent.java b/src/main/java/com/juick/xmpp/s2s/XMPPComponent.java
index 2b293fd6..13df5fd0 100644
--- a/src/main/java/com/juick/xmpp/s2s/XMPPComponent.java
+++ b/src/main/java/com/juick/xmpp/s2s/XMPPComponent.java
@@ -3,10 +3,16 @@ package com.juick.xmpp.s2s;
import com.juick.xmpp.Stanza;
import com.juick.xmpp.StanzaChild;
import com.juick.xmpp.extensions.JuickMessage;
+import org.xmlpull.v1.XmlPullParserException;
import javax.servlet.ServletContextEvent;
import javax.servlet.ServletContextListener;
import java.io.IOException;
+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;
@@ -100,11 +106,11 @@ public class XMPPComponent implements ServletContextListener {
return null;
}
- public static void sendOut(Stanza s) throws Exception {
+ public static void sendOut(Stanza s) {
sendOut(s.to.Host, s.toString());
}
- public static void sendOut(String hostname, String xml) throws Exception {
+ public static void sendOut(String hostname, String xml) {
boolean haveAnyConn = false;
ConnectionOut connOut = null;
@@ -146,8 +152,13 @@ public class XMPPComponent implements ServletContextListener {
}
if (!haveAnyConn) {
- ConnectionOut connectionOut = new ConnectionOut(hostname);
- XMPPComponent.executorService.submit(connectionOut);
+ ConnectionOut connectionOut = null;
+ try {
+ connectionOut = new ConnectionOut(hostname);
+ XMPPComponent.executorService.submit(connectionOut);
+ } catch (CertificateException | UnrecoverableKeyException | NoSuchAlgorithmException | XmlPullParserException | KeyStoreException | KeyManagementException | IOException e) {
+ LOGGER.log(Level.SEVERE, "s2s out error", e);
+ }
}
}
@@ -170,9 +181,7 @@ public class XMPPComponent implements ServletContextListener {
conf.getProperty("mysql_username", "") + "&password=" + conf.getProperty("mysql_password", ""));
childParsers.put(JuickMessage.XMLNS, new JuickMessage());
-
- connRouter = new ConnectionRouter(componentName);
- executorService.submit(connRouter);
+ executorService.submit(() -> connRouter = new ConnectionRouter(componentName, conf.getProperty("xmpp_password")));
executorService.submit(new ConnectionListener());
executorService.submit(new CleaningUp());
} catch (Exception e) {
@@ -201,7 +210,11 @@ public class XMPPComponent implements ServletContextListener {
}
}
- XMPPComponent.connRouter.closeConnection();
+ try {
+ XMPPComponent.connRouter.closeConnection();
+ } catch (IOException e) {
+ LOGGER.log(Level.WARNING, "router warning", e);
+ }
synchronized (XMPPComponent.sqlSync) {
if (XMPPComponent.sql != null) {