diff options
Diffstat (limited to 'src')
6 files changed, 654 insertions, 0 deletions
diff --git a/src/main/java/com/juick/ws/WebsocketComponent.java b/src/main/java/com/juick/ws/WebsocketComponent.java new file mode 100644 index 00000000..e87b96a5 --- /dev/null +++ b/src/main/java/com/juick/ws/WebsocketComponent.java @@ -0,0 +1,140 @@ +package com.juick.ws; + +import com.juick.User; +import com.juick.server.MessagesQueries; +import com.juick.server.UserQueries; +import org.apache.http.NameValuePair; +import org.apache.http.client.utils.URLEncodedUtils; +import org.springframework.http.HttpHeaders; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Component; +import org.springframework.web.socket.CloseStatus; +import org.springframework.web.socket.WebSocketSession; +import org.springframework.web.socket.handler.TextWebSocketHandler; + +import javax.inject.Inject; +import java.io.IOException; +import java.net.URI; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.logging.Level; +import java.util.logging.Logger; + +/** + * Created by vitalyster on 28.06.2016. + */ +@Component +public class WebsocketComponent extends TextWebSocketHandler { + + @Inject + JdbcTemplate jdbc; + + private static final Logger logger = Logger.getLogger(WebsocketComponent.class.getName()); + final List<SocketSubscribed> clients = Collections.synchronizedList(new ArrayList<SocketSubscribed>()); + + @Override + public void afterConnectionEstablished(WebSocketSession session) throws Exception { + URI hLocation; + String hXRealIP = ""; + + hLocation = session.getUri(); + HttpHeaders headers = session.getHandshakeHeaders(); + hXRealIP = headers.getOrDefault("X-Real-IP", + Collections.singletonList(session.getRemoteAddress().toString())).get(0); + + // Auth + User visitor = new User(); + List<NameValuePair> params = URLEncodedUtils.parse(hLocation, "UTF-8"); + for (NameValuePair param : params) { + if (param.getName().equals("hash")) { + String hash = param.getValue(); + if (hash.length() == 16) { + visitor = UserQueries.getUserByHash(jdbc, hash); + } else { + try { + logger.info(String.format("wrong hash for %d from %s", visitor.getUID(), hXRealIP)); + session.close(new CloseStatus(403, "Forbidden")); + } catch (IOException e) { + logger.log(Level.WARNING, "ws error", e); + } + } + break; + } + } + logger.info(String.format("user %d connected to %s from %s", visitor.getUID(), hLocation.getPath(), hXRealIP)); + + int MID = 0; + SocketSubscribed sockSubscr = null; + if (hLocation.getPath().equals("/")) { + logger.info(String.format("user %d connected", visitor.getUID())); + sockSubscr = new SocketSubscribed(session, hXRealIP, visitor, false); + } else if (hLocation.getPath().equals("/_all")) { + logger.info(String.format("user %d connected to legacy _all (%s)", visitor.getUID(), hLocation.getPath())); + sockSubscr = new SocketSubscribed(session, hXRealIP, visitor, true); + sockSubscr.allMessages = true; + } else if (hLocation.getPath().equals("/_replies")) { + logger.info(String.format("user %d connected to legacy _replies (%s)", visitor.getUID(), hLocation.getPath())); + sockSubscr = new SocketSubscribed(session, hXRealIP, visitor, true); + sockSubscr.allReplies = true; + } else if (hLocation.getPath().matches("/\\d+$")) { + try { + MID = Integer.parseInt(hLocation.getPath().substring(1)); + } catch (Exception e) { + } + if (MID > 0) { + if (MessagesQueries.canViewThread(jdbc, MID, visitor.getUID())) { + logger.info(String.format("user %d connected to legacy thread (%d) from %s", visitor.getUID(), MID, hXRealIP)); + sockSubscr = new SocketSubscribed(session, hXRealIP, visitor, true); + sockSubscr.MID = MID; + } else { + try { + session.close(new CloseStatus(403, "Forbidden")); + } catch (IOException e) { + logger.log(Level.WARNING, "ws error", e); + } + } + } + } + if (sockSubscr != null) { + synchronized (clients) { + clients.add(sockSubscr); + logger.info(clients.size() + " clients connected"); + } + } + } + @Override + public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception { + synchronized (clients) { + clients.removeIf(c -> { + if (c.session.getId().equals(session.getId())) { + logger.info(String.format("session from %s closed with status %d", c.clientName, status.getCode())); + return true; + } + return false; + }); + logger.info(clients.size() + " clients connected"); + } + + } + class SocketSubscribed { + + WebSocketSession session; + String clientName; + User visitor; + int MID; + boolean allMessages; + boolean allReplies; + long tsConnected; + long tsLastData; + boolean legacy; + + public SocketSubscribed(WebSocketSession session, String clientName, User visitor, boolean legacy) { + this.session = session; + this.clientName = clientName; + this.visitor = visitor; + tsConnected = tsLastData = System.currentTimeMillis(); + this.legacy = legacy; + } + } +} diff --git a/src/main/java/com/juick/ws/XMPPConnection.java b/src/main/java/com/juick/ws/XMPPConnection.java new file mode 100644 index 00000000..4a80eec5 --- /dev/null +++ b/src/main/java/com/juick/ws/XMPPConnection.java @@ -0,0 +1,165 @@ +package com.juick.ws; + +import com.juick.User; +import com.juick.json.MessageSerializer; +import com.juick.server.SubscriptionsQueries; +import com.juick.xmpp.JID; +import com.juick.xmpp.Message; +import com.juick.xmpp.Stream; +import com.juick.xmpp.StreamComponent; +import com.juick.xmpp.extensions.JuickMessage; +import org.springframework.core.env.Environment; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Component; +import org.springframework.web.socket.TextMessage; + +import javax.inject.Inject; +import java.io.IOException; +import java.net.Socket; +import java.util.List; +import java.util.logging.Level; +import java.util.logging.Logger; +import java.util.stream.Collectors; + +/** + * + * @author ugnich + */ +@Component +public class XMPPConnection implements Runnable, Stream.StreamListener, Message.MessageListener { + private static final Logger logger = Logger.getLogger(XMPPConnection.class.getName()); + @Inject + JdbcTemplate sql; + Stream xmpp; + String xmppPassword; + MessageSerializer ms; + WebsocketComponent wsHandler; + + @Inject + public XMPPConnection(Environment env, WebsocketComponent wsHandler) { + this.wsHandler = wsHandler; + xmppPassword = env.getProperty("xmpp_password"); + ms = new MessageSerializer(); + } + + @Override + public void run() { + try { + Socket socket = new Socket("localhost", 5347); + xmpp = new StreamComponent(new JID("", "ws.juick.com", ""), socket.getInputStream(), socket.getOutputStream(), xmppPassword); + xmpp.addChildParser(new JuickMessage()); + xmpp.addListener((Stream.StreamListener) this); + xmpp.addListener((Message.MessageListener) this); + xmpp.startParsing(); + } catch (IOException e) { + logger.log(Level.SEVERE, "XMPPConnection error", e); + } + } + + @Override + public void onStreamReady() { + logger.info("XMPP stream ready"); + } + + @Override + public void onStreamFail(Exception ex) { + logger.log(Level.SEVERE, "XMPP stream failed", ex); + } + + @Override + public void onMessage(com.juick.xmpp.Message msg) { + JuickMessage jmsg = (JuickMessage) msg.getChild(JuickMessage.XMLNS); + if (jmsg != null) { + logger.info("got jmsg: " + ms.serialize(jmsg).toString()); + if (jmsg.getMID() == 0) { + int uid_to = 0; + try { + uid_to = Integer.parseInt(msg.to.Username); + } catch (Exception e) { + } + if (uid_to > 0) { + onJuickPM(uid_to, jmsg); + } + } else if (jmsg.getRID() == 0) { + onJuickMessagePost(jmsg); + } else { + onJuickMessageReply(jmsg); + } + } + } + + MessageSerializer messageSerializer = new MessageSerializer(); + + private void onJuickPM(int uid_to, com.juick.Message jmsg) { + String json = messageSerializer.serialize(jmsg).toString(); + synchronized (wsHandler.clients) { + wsHandler.clients.stream().filter(c -> !c.legacy && c.visitor.getUID() == uid_to).forEach(c -> { + try { + logger.info("sending pm to " + c.visitor.getUID()); + c.session.sendMessage(new TextMessage(json)); + } catch (IOException e) { + logger.log(Level.WARNING, "ws error", e); + } + }); + } + } + + private void onJuickMessagePost(com.juick.Message jmsg) { + String json = messageSerializer.serialize(jmsg).toString(); + List<Integer> uids = SubscriptionsQueries.getSubscribedUsers(sql, jmsg.getUser().getUID(), jmsg.getMID()) + .stream().map(User::getUID).collect(Collectors.toList()); + synchronized (wsHandler.clients) { + wsHandler.clients.stream().filter(c -> + (!c.legacy && c.visitor.getUID() == 0) // anonymous users + || (!c.legacy && uids.contains(c.visitor.getUID()))) // subscriptions + .forEach(c -> { + try { + logger.info("sending message to " + c.visitor.getUID()); + c.session.sendMessage(new TextMessage(json)); + } catch (IOException e) { + logger.log(Level.WARNING, "ws error", e); + } + }); + wsHandler.clients.stream().filter(c -> + c.legacy && c.allMessages) // legacy all posts + .forEach(c -> { + try { + logger.info("sending message to legacy client " + c.visitor.getUID()); + c.session.sendMessage(new TextMessage(json)); + } catch (IOException e) { + logger.log(Level.WARNING, "ws error", e); + } + }); + } + } + + private void onJuickMessageReply(com.juick.Message jmsg) { + String json = messageSerializer.serialize(jmsg).toString(); + List<Integer> threadUsers = + SubscriptionsQueries.getUsersSubscribedToComments(sql, jmsg.getMID(), jmsg.getUser().getUID()) + .stream().map(User::getUID).collect(Collectors.toList()); + synchronized (wsHandler.clients) { + wsHandler.clients.stream().filter(c -> + (!c.legacy && c.visitor.getUID() == 0) // anonymous users + || (!c.legacy && threadUsers.contains(c.visitor.getUID()))) // subscriptions + .forEach(c -> { + try { + logger.info("sending reply to " + c.visitor.getUID()); + c.session.sendMessage(new TextMessage(json)); + } catch (IOException e) { + logger.log(Level.WARNING, "ws error", e); + } + }); + wsHandler.clients.stream().filter(c -> + (c.legacy && c.allReplies) || (c.legacy && c.MID == jmsg.getMID())) // legacy replies + .forEach(c -> { + try { + logger.info("sending reply to legacy client " + c.visitor.getUID()); + c.session.sendMessage(new TextMessage(json)); + } catch (IOException e) { + logger.log(Level.WARNING, "ws error", e); + } + }); + } + } +} diff --git a/src/main/java/com/juick/ws/configuration/WebsocketConfiguration.java b/src/main/java/com/juick/ws/configuration/WebsocketConfiguration.java new file mode 100644 index 00000000..223f8d63 --- /dev/null +++ b/src/main/java/com/juick/ws/configuration/WebsocketConfiguration.java @@ -0,0 +1,55 @@ +package com.juick.ws.configuration; + +import com.juick.ws.WebsocketComponent; +import com.juick.ws.XMPPConnection; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.ComponentScan; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.PropertySource; +import org.springframework.core.env.Environment; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DriverManagerDataSource; +import org.springframework.web.servlet.config.annotation.EnableWebMvc; +import org.springframework.web.servlet.config.annotation.WebMvcConfigurerAdapter; +import org.springframework.web.socket.config.annotation.EnableWebSocket; +import org.springframework.web.socket.config.annotation.WebSocketConfigurer; +import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry; + +import javax.inject.Inject; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +/** + * Created by vitalyster on 28.06.2016. + */ +@Configuration +@EnableWebMvc +@EnableWebSocket +@ComponentScan(basePackages = {"com.juick"}) +@PropertySource("classpath:juick.conf") +public class WebsocketConfiguration extends WebMvcConfigurerAdapter implements WebSocketConfigurer { + @Inject + Environment env; + ExecutorService xmppThread = Executors.newSingleThreadExecutor(); + @Bean + WebsocketComponent wsHandler() { + return new WebsocketComponent(); + } + @Bean + XMPPConnection xmpp() { + XMPPConnection xmpp = new XMPPConnection(env, wsHandler()); + xmppThread.submit(xmpp); + return xmpp; + } + @Bean + JdbcTemplate jdbc() { + DriverManagerDataSource dataSource = new DriverManagerDataSource(); + dataSource.setDriverClassName(env.getProperty("datasource_driver", "com.mysql.jdbc.Driver")); + dataSource.setUrl(env.getProperty("datasource_url")); + return new JdbcTemplate(dataSource); + } + @Override + public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { + registry.addHandler(wsHandler(), "/**").setAllowedOrigins("*"); + } +} diff --git a/src/main/java/com/juick/ws/configuration/WebsocketInitializer.java b/src/main/java/com/juick/ws/configuration/WebsocketInitializer.java new file mode 100644 index 00000000..89017f8b --- /dev/null +++ b/src/main/java/com/juick/ws/configuration/WebsocketInitializer.java @@ -0,0 +1,33 @@ +package com.juick.ws.configuration; +import org.springframework.web.filter.CharacterEncodingFilter; +import org.springframework.web.servlet.support.AbstractAnnotationConfigDispatcherServletInitializer; + +import javax.servlet.Filter; + +/** + * Created by vt on 09/02/16. + */ +public class WebsocketInitializer extends AbstractAnnotationConfigDispatcherServletInitializer { + @Override + protected Class<?>[] getRootConfigClasses() { + return new Class[] {WebsocketConfiguration.class}; + } + + @Override + protected Class<?>[] getServletConfigClasses() { + return null; + } + + @Override + protected String[] getServletMappings() { + return new String[] { + "/" + }; + } + @Override + protected Filter[] getServletFilters() { + CharacterEncodingFilter characterEncodingFilter = new CharacterEncodingFilter(); + characterEncodingFilter.setEncoding("UTF-8"); + return new Filter[] { characterEncodingFilter}; + } +} diff --git a/src/main/java/com/juick/xmpp/extensions/JuickMessage.java b/src/main/java/com/juick/xmpp/extensions/JuickMessage.java new file mode 100644 index 00000000..ac45f3d2 --- /dev/null +++ b/src/main/java/com/juick/xmpp/extensions/JuickMessage.java @@ -0,0 +1,186 @@ +/* + * Juick + * Copyright (C) 2008-2011, Ugnich Anton + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + */ +package com.juick.xmpp.extensions; + +import com.juick.xmpp.utils.XmlUtils; +import com.juick.xmpp.*; +import java.io.IOException; +import java.text.ParseException; +import java.text.SimpleDateFormat; +import java.util.TimeZone; + +import org.xmlpull.v1.XmlPullParser; +import org.xmlpull.v1.XmlPullParserException; + +/** + * + * @author Ugnich Anton + */ +public class JuickMessage extends com.juick.Message implements StanzaChild { + + public final static String XMLNS = "http://juick.com/message"; + public final static String TagName = "juick"; + + private SimpleDateFormat df; + + public JuickMessage() { + df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); + df.setTimeZone(TimeZone.getTimeZone("UTC")); + } + + public JuickMessage(com.juick.Message msg) { + super(msg); + df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); + df.setTimeZone(TimeZone.getTimeZone("UTC")); + } + + @Override + public String getXMLNS() { + return XMLNS; + } + + @Override + public JuickMessage parse(XmlPullParser parser) throws XmlPullParserException, IOException, ParseException { + JuickMessage jmsg = new JuickMessage(); + + final String sMID = parser.getAttributeValue(null, "mid"); + if (sMID != null) { + jmsg.setMID(Integer.parseInt(sMID)); + } + final String sRID = parser.getAttributeValue(null, "rid"); + if (sRID != null) { + jmsg.setRID(Integer.parseInt(sRID)); + } + final String sReplyTo = parser.getAttributeValue(null, "replyto"); + if (sReplyTo != null) { + jmsg.ReplyTo = Integer.parseInt(sReplyTo); + } + final String sPrivacy = parser.getAttributeValue(null, "privacy"); + if (sPrivacy != null) { + jmsg.Privacy = Integer.parseInt(sPrivacy); + } + final String sFriendsOnly = parser.getAttributeValue(null, "friendsonly"); + if (sFriendsOnly != null) { + jmsg.FriendsOnly = true; + } + final String sReadOnly = parser.getAttributeValue(null, "readonly"); + if (sReadOnly != null) { + jmsg.ReadOnly = true; + } + + String timestampString = parser.getAttributeValue(null, "ts"); + if (timestampString != null) { + jmsg.setDate(df.parse(timestampString)); + } + jmsg.AttachmentType = parser.getAttributeValue(null, "attach"); + + while (parser.next() == XmlPullParser.START_TAG) { + final String tag = parser.getName(); + final String xmlns = parser.getNamespace(); + if (tag.equals("body")) { + jmsg.setText(XmlUtils.getTagText(parser)); + } else if (tag.equals(JuickUser.TagName) && xmlns != null && xmlns.equals(JuickUser.XMLNS)) { + jmsg.setUser(new JuickUser().parse(parser)); + } else if (tag.equals("tag")) { + jmsg.Tags.add(XmlUtils.getTagText(parser)); + } else { + XmlUtils.skip(parser); + } + } + return jmsg; + } + + @Override + public String toString() { + String ret = ""; + + ret = "<" + TagName + " xmlns=\"" + XMLNS + "\""; + if (getMID() > 0) { + ret += " mid=\"" + getMID() + "\""; + } + if (getRID() > 0) { + ret += " rid=\"" + getRID() + "\""; + } + if (ReplyTo > 0) { + ret += " replyto=\"" + ReplyTo + "\""; + } + ret += " privacy=\"" + Privacy + "\""; + if (FriendsOnly) { + ret += " friendsonly=\"1\""; + } + if (ReadOnly) { + ret += " readonly=\"1\""; + } + if (getDate() != null) { + ret += " ts=\"" + df.format(getDate()) + "\""; + } + if (AttachmentType != null) { + ret += " attach=\"" + AttachmentType + "\""; + } + ret += ">"; + if (getUser() != null) { + ret += JuickUser.toString(getUser()); + } + if (getText() != null) { + ret += "<body>" + XmlUtils.escape(getText()) + "</body>"; + } + if (!Tags.isEmpty()) { + for (String Tag : Tags) { + ret += "<tag>" + XmlUtils.escape(Tag) + "</tag>"; + } + } + ret += "</" + TagName + ">"; + + return ret; + } + + @Override + public boolean equals(Object obj) { + if (!(obj instanceof JuickMessage)) { + return false; + } + JuickMessage jmsg = (JuickMessage) obj; + return (this.getMID() == jmsg.getMID() && this.getRID() == jmsg.getRID()); + } + + @Override + public int compareTo(Object obj) throws ClassCastException { + if (!(obj instanceof JuickMessage)) { + throw new ClassCastException(); + } + JuickMessage jmsg = (JuickMessage) obj; + + if (this.getMID() != jmsg.getMID()) { + if (this.getMID() > jmsg.getMID()) { + return -1; + } else { + return 1; + } + } + + if (this.getRID() != jmsg.getRID()) { + if (this.getRID() < jmsg.getRID()) { + return -1; + } else { + return 1; + } + } + + return 0; + } +} diff --git a/src/main/java/com/juick/xmpp/extensions/JuickUser.java b/src/main/java/com/juick/xmpp/extensions/JuickUser.java new file mode 100644 index 00000000..edc6749a --- /dev/null +++ b/src/main/java/com/juick/xmpp/extensions/JuickUser.java @@ -0,0 +1,75 @@ +/* + * Juick + * Copyright (C) 2008-2011, Ugnich Anton + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + */ +package com.juick.xmpp.extensions; + +import com.juick.xmpp.utils.XmlUtils; +import com.juick.xmpp.*; +import java.io.IOException; +import org.xmlpull.v1.XmlPullParser; +import org.xmlpull.v1.XmlPullParserException; + +/** + * + * @author Ugnich Anton + */ +public class JuickUser extends com.juick.User implements StanzaChild { + + public final static String XMLNS = "http://juick.com/user"; + public final static String TagName = "user"; + + public JuickUser() { + } + + public JuickUser(com.juick.User user) { + super(user); + } + + @Override + public String getXMLNS() { + return XMLNS; + } + + @Override + public JuickUser parse(final XmlPullParser parser) throws XmlPullParserException, IOException { + JuickUser juser = new JuickUser(); + String strUID = parser.getAttributeValue(null, "uid"); + if (strUID != null) { + juser.setUID(Integer.parseInt(strUID)); + } + juser.setUName(parser.getAttributeValue(null, "uname")); + XmlUtils.skip(parser); + return juser; + } + + public static String toString(com.juick.User user) { + String str = "<" + TagName + " xmlns='" + XMLNS + "'"; + if (user.getUID() > 0) { + str += " uid='" + user.getUID() + "'"; + } + if (user.getUName() != null && user.getUName().length() > 0) { + str += " uname='" + XmlUtils.escape(user.getUName()) + "'"; + } + str += "/>"; + return str; + } + + @Override + public String toString() { + return toString(this); + } +} |