From 2a8c4f613f397619c563fbb74c04e39074d6e98e Mon Sep 17 00:00:00 2001 From: Ugnich Anton Date: Wed, 14 Aug 2013 09:21:47 +0700 Subject: RFC 6455 --- src/com/juick/jabber/ws/Main.java | 10 +- src/com/juick/jabber/ws/SocketSubscribed.java | 15 +- src/com/juick/jabber/ws/WSConnections.java | 4 +- src/com/juick/jabber/ws/WSData.java | 337 +++++++++++++++----------- src/com/juick/jabber/ws/WSKeepAlive.java | 106 ++++++++ src/com/juick/jabber/ws/XMPPConnection.java | 179 +++++--------- 6 files changed, 379 insertions(+), 272 deletions(-) (limited to 'src/com/juick/jabber') diff --git a/src/com/juick/jabber/ws/Main.java b/src/com/juick/jabber/ws/Main.java index 7036b0d4..a7aea543 100644 --- a/src/com/juick/jabber/ws/Main.java +++ b/src/com/juick/jabber/ws/Main.java @@ -32,9 +32,8 @@ public class Main { Connection sql; XMPPConnection xmpp; - public final static ArrayList sockReplies = new ArrayList(); - public final static ArrayList sockMessages = new ArrayList(); - public final static ArrayList sockAll = new ArrayList(); + public static String STATSFILE = null; + public final static ArrayList clients = new ArrayList(); public static void main(String[] args) { new Main().start(); @@ -45,11 +44,14 @@ public class Main { Properties conf = new Properties(); conf.load(new FileInputStream("/etc/juick/ws.conf")); + STATSFILE = conf.getProperty("statsfile"); + setupSql(conf.getProperty("mysql_username", ""), conf.getProperty("mysql_password", "")); xmpp = new XMPPConnection(sql, conf.getProperty("xmpp_password", "")); new Thread(xmpp).start(); + + //new Thread(new WSConnections()).start(); new Thread(new WSData(sql)).start(); - new Thread(new WSConnections()).start(); new Thread(new WSKeepAlive(sql)).start(); } catch (Exception e) { System.err.println(e); diff --git a/src/com/juick/jabber/ws/SocketSubscribed.java b/src/com/juick/jabber/ws/SocketSubscribed.java index 29baf0b8..f865ce6a 100644 --- a/src/com/juick/jabber/ws/SocketSubscribed.java +++ b/src/com/juick/jabber/ws/SocketSubscribed.java @@ -9,14 +9,19 @@ import java.nio.channels.SocketChannel; public class SocketSubscribed { public SocketChannel sock = null; + public String clientName = null; + public int VUID = 0; public int UID = 0; public int MID = 0; - public long tsConnected = 0; - public long tsLastData = 0; + public boolean allMessages = false; + public boolean allReplies = false; + public long tsConnected; + public long tsLastData; - public SocketSubscribed(SocketChannel sock, int UID, int MID) { + public SocketSubscribed(SocketChannel sock, String clientName, int VUID) { this.sock = sock; - this.UID = UID; - this.MID = MID; + this.clientName = clientName; + this.VUID = VUID; + tsConnected = tsLastData = System.currentTimeMillis(); } } diff --git a/src/com/juick/jabber/ws/WSConnections.java b/src/com/juick/jabber/ws/WSConnections.java index 020442f7..15fbe4e8 100644 --- a/src/com/juick/jabber/ws/WSConnections.java +++ b/src/com/juick/jabber/ws/WSConnections.java @@ -32,9 +32,9 @@ public class WSConnections implements Runnable { it.remove(); ServerSocketChannel ssChannel = (ServerSocketChannel) selKey.channel(); SocketChannel sChannel = ssChannel.accept(); - System.out.println(sChannel.socket().getRemoteSocketAddress().toString() + " ACCEPTED"); sChannel.configureBlocking(false); - sChannel.register(WSData.sel, SelectionKey.OP_READ); + sChannel.register(sel, SelectionKey.OP_READ); + System.out.println(sChannel.socket().getRemoteSocketAddress().toString() + " ACCEPTED"); } } } catch (Exception e) { diff --git a/src/com/juick/jabber/ws/WSData.java b/src/com/juick/jabber/ws/WSData.java index 97193510..d77257e6 100644 --- a/src/com/juick/jabber/ws/WSData.java +++ b/src/com/juick/jabber/ws/WSData.java @@ -1,14 +1,19 @@ package com.juick.jabber.ws; +import com.juick.server.MessagesQueries; +import com.juick.server.UserQueries; +import com.juick.xmpp.utils.Base64; import java.io.IOException; -import java.math.BigInteger; +import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.CharBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; +import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; import java.nio.charset.Charset; import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; import java.sql.Connection; import java.util.Iterator; @@ -17,46 +22,77 @@ import java.util.Iterator; * @author ugnich */ public class WSData implements Runnable { - + + static final String WEBSOCKET_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; Connection sql; - public static Selector sel; - + public Selector sel; + public WSData(Connection sql) { this.sql = sql; } - + @Override public void run() { try { sel = Selector.open(); + ServerSocketChannel listensock = ServerSocketChannel.open(); + listensock.configureBlocking(false); + listensock.socket().bind(new InetSocketAddress(8081)); + listensock.register(sel, SelectionKey.OP_ACCEPT); + while (true) { sel.select(); Iterator it = sel.selectedKeys().iterator(); while (it.hasNext()) { SelectionKey selKey = it.next(); it.remove(); - - SocketChannel sChannel = (SocketChannel) selKey.channel(); - ByteBuffer buf = ByteBuffer.allocate(10240); - try { - if (sChannel.read(buf) > 0) { - buf.flip(); - CharBuffer charbuf = Charset.forName("ISO-8859-1").decode(buf); - if (charbuf.charAt(0) == 0 && charbuf.charAt(charbuf.length() - 1) == 0xFF) { - wsTextFrame(sChannel, charbuf.subSequence(1, charbuf.length() - 2)); - } else if (charbuf.charAt(0) == 'G' && charbuf.charAt(1) == 'E' && charbuf.charAt(2) == 'T' && charbuf.charAt(3) == ' ') { - wsHandshake(sChannel, buf); - } else { - throw new IOException(sChannel.socket().getRemoteSocketAddress().toString() + " INVALID FRAME"); + + if (selKey.isAcceptable()) { + ServerSocketChannel ssChannel = (ServerSocketChannel) selKey.channel(); + SocketChannel sChannel = ssChannel.accept(); + sChannel.configureBlocking(false); + sChannel.register(sel, SelectionKey.OP_READ); + System.out.println(sChannel.socket().getRemoteSocketAddress().toString() + " ACCEPTED"); + } else if (selKey.isReadable()) { + SocketChannel sChannel = (SocketChannel) selKey.channel(); + ByteBuffer buf = ByteBuffer.allocate(10240); + try { + int readbytes = sChannel.read(buf); + if (readbytes > 0) { + buf.flip(); + + CharBuffer charbuf = Charset.forName("ISO-8859-1").decode(buf); + System.out.println("DATA: " + charbuf.toString()); + buf.rewind(); + + switch (buf.get(0)) { + case (byte) 0x89: // PING + updateSocketTS(sChannel); + wsPing(sChannel); + break; + case (byte) 0x8A: // PONG + updateSocketTS(sChannel); + break; + case (byte) 0x81: // TEXT FRAME + updateSocketTS(sChannel); + wsTextFrame(sChannel, buf); + break; + case (byte) 'G': // HTTP + updateSocketTS(sChannel); + wsHandshake(sChannel, buf); + break; + case (byte) 0x88: // CONNECTION CLOSE + throw new IOException(sChannel.socket().getRemoteSocketAddress().toString() + " CONNECTION CLOSE"); + } + } else if (readbytes < 0) { + throw new IOException(sChannel.socket().getRemoteSocketAddress().toString() + " END OF STREAM"); } - } else { - throw new IOException(sChannel.socket().getRemoteSocketAddress().toString()+ " NO DATA"); + } catch (IOException e) { + System.err.println("WSData: " + e); + sChannel.socket().close(); + sChannel.close(); + selKey.cancel(); } - } catch (IOException e) { - System.err.println("WSData: " + e); - sChannel.socket().close(); - sChannel.close(); - selKey.cancel(); } } } @@ -64,15 +100,15 @@ public class WSData implements Runnable { System.err.println("WSData: " + e); } } - + public void wsHandshake(SocketChannel sock, ByteBuffer buf) throws Exception { String hOrigin = null; String hHost = null; String hLocation = null; - String hSecWebSocketKey1 = null; - String hSecWebSocketKey2 = null; - String hCookie = null; - + String hSecWebSocketKey = null; + String hSecWebSocketVersion = null; + String hXRealIP = null; + buf.rewind(); CharBuffer charbuf = Charset.forName("ISO-8859-1").decode(buf); String headers[] = charbuf.toString().split("\r\n"); @@ -85,149 +121,156 @@ public class WSData implements Runnable { hOrigin = h[1]; } else if (h[0].equals("Host:")) { hHost = h[1]; - } else if (h[0].equals("Sec-WebSocket-Key1:")) { - hSecWebSocketKey1 = h[1]; - } else if (h[0].equals("Sec-WebSocket-Key2:")) { - hSecWebSocketKey2 = h[1]; - } else if (h[0].equals("Cookie:")) { - hCookie = h[1]; + } else if (h[0].equals("Sec-WebSocket-Key:")) { + hSecWebSocketKey = h[1]; + } else if (h[0].equals("Sec-WebSocket-Version:")) { + hSecWebSocketVersion = h[1]; + } else if (h[0].equals("X-Real-IP:")) { + hXRealIP = h[1]; } } } - - if (hOrigin == null || hHost == null || hLocation == null || hSecWebSocketKey1 == null || hSecWebSocketKey2 == null) { + + if (hOrigin == null || hHost == null || hLocation == null || hSecWebSocketKey == null || hSecWebSocketVersion == null || !hSecWebSocketVersion.equals("13")) { throw new IOException(sock.socket().getRemoteSocketAddress().toString() + " Invalid headers"); } - // Cookies - int UID = 0; - String hash = null; - if (hCookie != null) { - String cookies[] = hCookie.split("; "); - for (int i = 0; i < cookies.length; i++) { - String cookie[] = cookies[i].split("=", 2); - if (cookie[0].equals("hash")) { - hash = cookie[1]; - break; - } + // Auth + int VUID = 0; + int hashloc = hLocation.indexOf("hash="); + if (hashloc > 0) { + String hash = hLocation.substring(hashloc + 5); + if (hash.indexOf('&') > 0) { + hash = hash.substring(0, hash.indexOf('&')); } - if (hash != null) { - UID = com.juick.server.UserQueries.getUIDbyHash(sql, hash); + if (hash.length() == 16) { + VUID = com.juick.server.UserQueries.getUIDbyHash(sql, hash); } } // URL - String loc[] = hLocation.split("/"); + int hLocationQM = hLocation.indexOf('?'); + if (hLocationQM > 0) { + hLocation = hLocation.substring(0, hLocationQM); + } + int MID = 0; - if (hLocation.equals("/my") && UID > 0) { - Main.sockMessages.add(new SocketSubscribed(sock, UID, 0)); - } else if (hLocation.equals("/all")) { - Main.sockAll.add(new SocketSubscribed(sock, UID, 0)); - } else if ((loc.length == 2 || loc.length == 3) && loc[1].equals("replies")) { - if (loc.length == 2) { - Main.sockReplies.add(new SocketSubscribed(sock, UID, 0)); - } else { - try { - MID = Integer.parseInt(loc[2]); - } catch (Exception e) { - } - if (MID > 0) { - Main.sockReplies.add(new SocketSubscribed(sock, UID, MID)); + int responseCode = 404; + SocketSubscribed sockSubscr = null; + if (hLocation.equals("/") && VUID > 0) { + sockSubscr = new SocketSubscribed(sock, hXRealIP, VUID); + responseCode = 101; + } else if (hLocation.equals("/_all")) { + sockSubscr = new SocketSubscribed(sock, hXRealIP, VUID); + sockSubscr.allMessages = true; + responseCode = 101; + } else if (hLocation.equals("/_replies")) { + sockSubscr = new SocketSubscribed(sock, hXRealIP, VUID); + sockSubscr.allReplies = true; + responseCode = 101; + } else if (hLocation.matches("^/\\d+$")) { + try { + MID = Integer.parseInt(hLocation.substring(1)); + } catch (Exception e) { + } + if (MID > 0) { + if (MessagesQueries.canViewThread(sql, MID, VUID)) { + sockSubscr = new SocketSubscribed(sock, hXRealIP, VUID); + sockSubscr.MID = MID; + responseCode = 101; } else { - throw new IOException(sock.socket().getRemoteSocketAddress().toString() + " Invalid MID"); + responseCode = 403; } } - } else { - throw new IOException(sock.socket().getRemoteSocketAddress().toString() + " Invalid location"); - } - - System.out.println(sock.socket().getRemoteSocketAddress().toString() + " HANDSHAKE (Hash=" + hash + "; UID = " + UID + "; MID = " + MID + ")"); - - Long lSecNum1 = calcSecKeyNum(hSecWebSocketKey1); - Long lSecNum2 = calcSecKeyNum(hSecWebSocketKey2); - - BigInteger sec1 = new BigInteger(lSecNum1.toString()); - BigInteger sec2 = new BigInteger(lSecNum2.toString()); - - // concatenate 3 parts secNum1 + secNum2 + secKey (16 Bytes) - byte[] l128Bit = new byte[]{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}; - byte[] lTmp; - - lTmp = sec1.toByteArray(); - int lIdx = lTmp.length; - int lCnt = 0; - while (lIdx > 0 && lCnt < 4) { - lIdx--; - lCnt++; - l128Bit[4 - lCnt] = lTmp[lIdx]; + } else if (hLocation.matches("^/[a-zA-Z0-9\\-]{2,16}/?$")) { + String uname; + if (hLocation.endsWith("/")) { + uname = hLocation.substring(1, hLocation.length() - 2); + } else { + uname = hLocation.substring(1); + } + + int UID = UserQueries.getUIDbyName(sql, uname); + if (UID > 0) { + // check access + sockSubscr = new SocketSubscribed(sock, hXRealIP, VUID); + sockSubscr.UID = UID; + responseCode = 101; + } } - - lTmp = sec2.toByteArray(); - lIdx = lTmp.length; - lCnt = 0; - while (lIdx > 0 && lCnt < 4) { - lIdx--; - lCnt++; - l128Bit[8 - lCnt] = lTmp[lIdx]; + if (sockSubscr != null) { + synchronized (Main.clients) { + Main.clients.add(sockSubscr); + } } - buf.rewind(); - for (int i = 0; i < 8; i++) { - l128Bit[8 + i] = buf.get(buf.limit() - 8 + i); + // Response + String outstr; + if (responseCode == 101) { + outstr = "HTTP/1.1 101 Switching Protocols\r\n" + + "Upgrade: websocket\r\n" + + "Connection: Upgrade\r\n" + + "Sec-WebSocket-Accept: " + calcHeaderAccept(hSecWebSocketKey) + "\r\n" + + "\r\n"; + } else if (responseCode == 403) { + outstr = "HTTP/1.1 403 Forbidden\r\n\r\n"; + } else { + outstr = "HTTP/1.1 404 Not Found\r\n\r\n"; } - - String outstr = "HTTP/1.1 101 Web Socket Protocol Handshake\r\n" - + "Upgrade: WebSocket\r\n" - + "Connection: Upgrade\r\n" - + "Sec-WebSocket-Origin: " + hOrigin + "\r\n" - + "Sec-WebSocket-Location: ws://" + hHost + hLocation + "\r\n" - + "Sec-WebSocket-Protocol: sample\r\n" - + "\r\n"; - ByteBuffer out = ByteBuffer.allocate(4096); + ByteBuffer out = ByteBuffer.allocate(1024); out.put(Charset.forName("ISO-8859-1").encode(outstr)); - out.put(MessageDigest.getInstance("MD5").digest(l128Bit)); out.flip(); - sock.write(out); - } - - private static long calcSecKeyNum(String aKey) { - StringBuilder lSB = new StringBuilder(); - int lSpaces = 0; - for (int i = 0; i < aKey.length(); i++) { - char lC = aKey.charAt(i); - if (lC == ' ') { - lSpaces++; - } else if (lC >= '0' && lC <= '9') { - lSB.append(lC); - } + + if (responseCode == 101) { + System.out.println(sock.socket().getRemoteSocketAddress().toString() + " HANDSHAKE (VUID = " + VUID + "; MID = " + MID + ")"); + } else { + throw new IOException(sock.socket().getRemoteSocketAddress().toString() + " " + responseCode); } - long lRes = -1; - if (lSpaces > 0) { - try { - lRes = Long.parseLong(lSB.toString()) / lSpaces; - } catch (NumberFormatException ex) { - // use default result - } + } + + private String calcHeaderAccept(String key) { + String base = key + WEBSOCKET_GUID; + try { + MessageDigest md = MessageDigest.getInstance("SHA-1"); + return Base64.encode(md.digest(base.getBytes())); + } catch (NoSuchAlgorithmException e) { + System.err.println("calcHeaderAccept: " + e); } - return lRes; + return ""; } - - public void wsTextFrame(SocketChannel sock, CharSequence csbuf) { - String buf = csbuf.toString(); - if (buf.equals(" ")) { - ByteBuffer out = ByteBuffer.allocate(4); - out.put((byte) 0x00); - out.put((byte) 0x20); - out.put((byte) 0xFF); - out.flip(); - out.rewind(); - try { - sock.write(out); - } catch (IOException e) { + + public void wsPing(SocketChannel sock) throws Exception { + ByteBuffer out = ByteBuffer.allocate(2); + out.put((byte) 0x8A); // PONG FRAME + out.put((byte) 0x00); // 1 byte long + out.flip(); + out.rewind(); + sock.write(out); + } + + public void wsTextFrame(SocketChannel sock, ByteBuffer buf) throws Exception { + /* + ByteBuffer out = ByteBuffer.allocate(3); + out.put((byte) 0x81); // TEXT FRAME + out.put((byte) 0x01); // 1 byte long + out.put((byte) 0x20); // ' ' + out.flip(); + out.rewind(); + sock.write(out); + */ + } + + public void updateSocketTS(SocketChannel sock) { + synchronized (Main.clients) { + Iterator i = Main.clients.iterator(); + while (i.hasNext()) { + SocketSubscribed s = i.next(); + if (s.sock == sock) { + s.tsLastData = System.currentTimeMillis(); + break; + } } - } else { - System.out.println(sock.socket().getRemoteSocketAddress().toString() + " DATA '" + buf + "'"); } } } diff --git a/src/com/juick/jabber/ws/WSKeepAlive.java b/src/com/juick/jabber/ws/WSKeepAlive.java index ba6376a4..2e36b100 100644 --- a/src/com/juick/jabber/ws/WSKeepAlive.java +++ b/src/com/juick/jabber/ws/WSKeepAlive.java @@ -1,6 +1,9 @@ package com.juick.jabber.ws; +import java.io.PrintWriter; +import java.nio.ByteBuffer; import java.sql.Connection; +import java.util.Iterator; /** * @@ -9,12 +12,115 @@ import java.sql.Connection; public class WSKeepAlive implements Runnable { Connection sql; + ByteBuffer pingBytes; + ByteBuffer closeBytes; public WSKeepAlive(Connection sql) { this.sql = sql; + + //pingBytes = ByteBuffer.allocate(2); + //pingBytes.put((byte) 0x8A); // PONG FRAME + //pingBytes.put((byte) 0x00); // 0 byte long + pingBytes = ByteBuffer.allocate(3); + pingBytes.put((byte) 0x81); // TEXT FRAME + pingBytes.put((byte) 0x01); // 1 byte long + pingBytes.put((byte) 0x20); // ' ' + pingBytes.flip(); + + closeBytes = ByteBuffer.allocate(2); + closeBytes.put((byte) 0x88); // CLOSE FRAME + closeBytes.put((byte) 0x00); // 0 byte long + closeBytes.flip(); } @Override public void run() { + while (true) { + PrintWriter statsFile = null; + + if (Main.STATSFILE != null) { + try { + statsFile = new PrintWriter(Main.STATSFILE, "UTF-8"); + } catch (Exception e) { + statsFile = null; + System.err.println("WSKeepAlive statsFile: " + e); + } + } + + long now = System.currentTimeMillis(); + + synchronized (Main.clients) { + if (statsFile != null) { + statsFile.write("

Connections (" + Main.clients.size() + ")

"); + } + + for (Iterator i = Main.clients.iterator(); i.hasNext();) { + SocketSubscribed s = i.next(); + int inactive = (int) ((double) (now - s.tsLastData) / 1000.0); + + if (statsFile != null) { + try { + statsFile.print(""); + statsFile.print(""); + statsFile.print(""); + statsFile.print(""); + statsFile.print(""); + statsFile.print(""); + statsFile.print(""); + } catch (Exception e) { + System.err.println("WSKeepAlive statsFile print: " + e); + } + } + + if (inactive > 180) { + closeBytes.rewind(); + try { + s.sock.write(closeBytes); + } catch (Exception e) { + } finally { + try { + s.sock.socket().close(); + } catch (Exception ex) { + } + try { + s.sock.close(); + } catch (Exception ex) { + } + i.remove(); + } + } else if (inactive > 60) { + pingBytes.rewind(); + try { + s.sock.write(pingBytes); + } catch (Exception e) { + System.err.println("WSKeepAlive ping: " + e); + try { + s.sock.socket().close(); + } catch (Exception ex) { + } + try { + s.sock.close(); + } catch (Exception ex) { + } + i.remove(); + } + } + } + } + + if (Main.STATSFILE != null) { + try { + statsFile.write("
IPinactiveVUIDUIDMIDallMallR
" + (s.clientName != null ? s.clientName : "?") + "" + inactive + "" + (s.VUID > 0 ? s.VUID : "") + "" + (s.UID > 0 ? s.UID : "") + "" + (s.MID > 0 ? s.MID : "") + "" + (s.allMessages ? "+" : "") + "" + (s.allReplies ? "+" : "") + "
"); + statsFile.close(); + } catch (Exception e) { + System.err.println("WSKeepAlive statsFile close: " + e); + } + } + + try { + Thread.sleep(10000); + } catch (InterruptedException e) { + } + } } } diff --git a/src/com/juick/jabber/ws/XMPPConnection.java b/src/com/juick/jabber/ws/XMPPConnection.java index 91a8387b..24329992 100644 --- a/src/com/juick/jabber/ws/XMPPConnection.java +++ b/src/com/juick/jabber/ws/XMPPConnection.java @@ -1,5 +1,7 @@ package com.juick.jabber.ws; +import com.juick.server.MessagesQueries; +import com.juick.server.Utils; import com.juick.xmpp.JID; import com.juick.xmpp.Message; import com.juick.xmpp.Stream; @@ -13,6 +15,8 @@ import java.sql.Connection; import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Statement; +import java.util.ArrayList; +import java.util.Iterator; /** * @@ -33,7 +37,7 @@ public class XMPPConnection implements Runnable, Stream.StreamListener, Message. public void run() { try { Socket socket = new Socket("localhost", 5347); - xmpp = new StreamComponent(new JID("", "ws.juick.com", ""), socket.getInputStream(), socket.getOutputStream(), xmppPassword); + xmpp = new StreamComponent(new JID("", "ws2.juick.com", ""), socket.getInputStream(), socket.getOutputStream(), xmppPassword); xmpp.addChildParser(new JuickMessage()); xmpp.addListener((Stream.StreamListener) this); xmpp.addListener((Message.MessageListener) this); @@ -67,35 +71,24 @@ public class XMPPConnection implements Runnable, Stream.StreamListener, Message. } private void onJuickMessagePost(com.juick.Message jmsg) { - String json = "{" - + "\"mid\":" + jmsg.MID + "," - + "\"user\":{" + "\"uid\":" + jmsg.User.UID + "," + "\"uname\":\"" + encloseJSON(jmsg.User.UName) + "\"}," - + "\"timestamp\":\"" + jmsg.TimestampString + "\"," - + "\"body\":\"" + encloseJSON(jmsg.Text) + "\""; - if (jmsg.Tags.size() > 0) { - json += ",\"tags\":["; - for (int i = 0; i < jmsg.Tags.size(); i++) { - if (i > 0) { - json += ","; - } - json += "\"" + encloseJSON((String) jmsg.Tags.get(i)) + "\""; - } - json += "]"; + String json = com.juick.json.Message.toJSON(jmsg).toString(); + ByteBuffer jsonbytes = Charset.forName("UTF-8").encode(json); + ByteBuffer bbMsg = ByteBuffer.allocate(10240); + bbMsg.put((byte) 0x81); + if (jsonbytes.limit() <= 125) { + bbMsg.put((byte) jsonbytes.limit()); + } else { + bbMsg.put((byte) 126); + bbMsg.putShort((short) jsonbytes.limit()); } - json += "}"; - - ByteBuffer out = ByteBuffer.allocate(10240); - out.put((byte) 0x00); - out.put(Charset.forName("UTF-8").encode(json)); - out.put((byte) 0xFF); - out.flip(); - + bbMsg.put(jsonbytes); + bbMsg.flip(); + ArrayList uids = new ArrayList(); String query = "SELECT suser_id FROM subscr_users WHERE user_id=" + jmsg.User.UID + " AND suser_id NOT IN (SELECT user_id FROM bl_tags INNER JOIN messages_tags USING(tag_id) WHERE message_id=" + jmsg.MID + ")"; if (jmsg.Privacy < 0) { query += " AND suser_id IN (SELECT wl_user_id FROM wl_users WHERE user_id=" + jmsg.User.UID + ")"; } - Statement stmt = null; ResultSet rs = null; try { @@ -103,115 +96,73 @@ public class XMPPConnection implements Runnable, Stream.StreamListener, Message. rs = stmt.executeQuery(query); rs.beforeFirst(); while (rs.next()) { - int UID = rs.getInt(1); - - for (int i = Main.sockMessages.size() - 1; i >= 0; i--) { - SocketSubscribed ss = Main.sockMessages.get(i); - if (ss.UID == UID) { - try { - out.rewind(); - ss.sock.write(out); - } catch (IOException e) { - Main.sockMessages.remove(i); - try { - ss.sock.close(); - } catch (IOException ex) { - } - } - } - } - - if (jmsg.Privacy <= 0) { - for (int i = Main.sockAll.size() - 1; i >= 0; i--) { - SocketSubscribed ss = Main.sockAll.get(i); - if (ss.UID == UID) { - try { - out.rewind(); - ss.sock.write(out); - } catch (IOException e) { - Main.sockAll.remove(i); - try { - ss.sock.close(); - } catch (IOException ex) { - } - } - } - } - } - + uids.add(rs.getInt(1)); } } catch (SQLException e) { - System.err.println(e); + System.err.println("onJuickMessagePost: " + e); } finally { - if (rs != null) { - try { - rs.close(); - } catch (SQLException e) { - } - } - if (stmt != null) { - try { - stmt.close(); - } catch (SQLException e) { - } - } + Utils.finishSQL(rs, stmt); } - // Send to all - if (jmsg.Privacy > 0) { - for (int i = Main.sockAll.size() - 1; i >= 0; i--) { - SocketSubscribed ss = Main.sockAll.get(i); - try { - out.rewind(); - ss.sock.write(out); - System.err.println(" --->" + ss.sock.socket().getRemoteSocketAddress()); - } catch (IOException e) { - Main.sockAll.remove(i); + synchronized (Main.clients) { + for (Iterator i = Main.clients.iterator(); i.hasNext();) { + SocketSubscribed s = i.next(); + if ((jmsg.Privacy >= 0 && (s.allMessages || s.UID == jmsg.User.UID)) || uids.contains(s.VUID)) { + bbMsg.rewind(); try { - ss.sock.close(); - } catch (IOException ex) { + s.sock.write(bbMsg); + } catch (Exception e) { + try { + s.sock.socket().close(); + } catch (Exception ex) { + } + try { + s.sock.close(); + } catch (Exception ex) { + } + i.remove(); } } } } - } private void onJuickMessageReply(com.juick.Message jmsg) { - String json = "{" - + "\"mid\":" + jmsg.MID + "," - + "\"rid\":" + jmsg.RID + "," - + "\"replyto\":" + jmsg.ReplyTo + "," - + "\"user\":{" + "\"uid\":" + jmsg.User.UID + "," + "\"uname\":\"" + encloseJSON(jmsg.User.UName) + "\"}," - + "\"timestamp\":\"" + jmsg.TimestampString + "\"," - + "\"body\":\"" + encloseJSON(jmsg.Text) + "\"" - + "}"; + String json = com.juick.json.Message.toJSON(jmsg).toString(); + ByteBuffer jsonbytes = Charset.forName("UTF-8").encode(json); + ByteBuffer bbMsg = ByteBuffer.allocate(10240); + bbMsg.put((byte) 0x81); + if (jsonbytes.limit() <= 125) { + bbMsg.put((byte) jsonbytes.limit()); + } else { + bbMsg.put((byte) 126); + bbMsg.putShort((short) jsonbytes.limit()); + } + bbMsg.put(jsonbytes); + bbMsg.flip(); - ByteBuffer out = ByteBuffer.allocate(10240); - out.put((byte) 0x00); - out.put(Charset.forName("UTF-8").encode(json)); - out.put((byte) 0xFF); - out.flip(); + int privacy = MessagesQueries.getMessagePrivacy(sql, jmsg.MID); - for (int i = Main.sockReplies.size() - 1; i >= 0; i--) { - SocketSubscribed ss = Main.sockReplies.get(i); - if (ss.MID == 0 || ss.MID == jmsg.MID) { - try { - out.rewind(); - ss.sock.write(out); - System.err.println(" --->" + ss.sock.socket().getRemoteSocketAddress()); - } catch (IOException e) { - Main.sockReplies.remove(i); + synchronized (Main.clients) { + for (Iterator i = Main.clients.iterator(); i.hasNext();) { + SocketSubscribed s = i.next(); + if ((privacy >= 0 && s.allReplies) || s.MID == jmsg.MID) { + bbMsg.rewind(); try { - ss.sock.close(); - } catch (IOException ex) { + s.sock.write(bbMsg); + } catch (Exception e) { + try { + s.sock.socket().close(); + } catch (Exception ex) { + } + try { + s.sock.close(); + } catch (Exception ex) { + } + i.remove(); } } } } } - - public static String encloseJSON(String str) { - return str.replace("\"", """).replace("\\", "\\\\").replace("\n", "\\n"); - } } -- cgit v1.2.3