Merge branch 'surevine-issues/213'

Author: abmargb

postgres/upgrade-7.sql

@@ -0,0 +1,9 @@
+BEGIN TRANSACTION;
+
+CREATE TABLE "online_users" ("user" TEXT NOT NULL,
+			  "updated" TIMESTAMP);
+
+INSERT INTO schema_version (version, "when", description)
+       VALUES (7, 'now', 'Added presence table');
+
+COMMIT;

src/main/java/org/buddycloud/channelserver/ChannelsEngine.java

@@ -113,7 +113,7 @@ private void setupManagers() throws ComponentException {
 				nodeStoreFactory);
 		federatedQueueManager = new FederatedQueueManager(this,
 				configuration);
-		onlineUsers = new OnlineResourceManager(configuration);
+		onlineUsers = new OnlineResourceManager(configuration, channelManagerFactory.create());
 	}
 
 	@Override

src/main/java/org/buddycloud/channelserver/Configuration.java

@@ -36,12 +36,13 @@ public class Configuration extends Properties {
 
 	public static final String DISCOVERY_USE_DNS = "discovery.dns.enabled";
 
+	public static final String PERSIST_PRESENCE_DATA = "users.presence.persist";
+	
 	public static final String NOTIFICATIONS_SENDTO = "notifications.sendTo";
 	public static final String NOTIFICATIONS_CONNECTED = "notifications.connected";
 
 	private static final String CONFIGURATION_FILE = "configuration.properties";
 
-	public static final String PURGE_REMOTE_ON_START = "sync.purge-on-start";
 
 	private static Configuration instance = null;
 

src/main/java/org/buddycloud/channelserver/channel/ChannelManagerImpl.java

@@ -448,4 +448,19 @@ public int getCountUserFeedItems(JID user, Date since,
 		return nodeStore.getCountUserFeedItems(user, since, parentOnly);
 	}
 
+	@Override
+	public void jidOnline(JID jid) throws NodeStoreException {
+		nodeStore.jidOnline(jid);
+	}
+
+	@Override
+	public void jidOffline(JID jid) throws NodeStoreException {
+		nodeStore.jidOffline(jid);
+	}
+
+	@Override
+	public ArrayList<JID> onlineJids(JID jid) throws NodeStoreException {
+		return nodeStore.onlineJids(jid);
+	}
+
 }

src/main/java/org/buddycloud/channelserver/db/NodeStore.java

@@ -595,6 +595,31 @@ boolean userHasRatedPost(String node, JID user, GlobalItemID id)
 	 */
 	ResultSet<NodeThread> getNodeThreads(String node, String afterId, int limit)
 			throws NodeStoreException;
+	
+	/**
+	 * A JID has come online
+	 * 
+	 * @param jid
+	 * @throws NodeStoreException 
+	 */
+	void jidOnline(JID jid) throws NodeStoreException;
+	
+	/**
+	 * A JID has gone offline
+	 * 
+	 * @param jid
+	 * @throws NodeStoreException 
+	 */
+	void jidOffline(JID jid) throws NodeStoreException;
+	
+	/**
+	 * Get online resources
+	 * @return 
+	 * @throws NodeStoreException 
+	 * 
+	 * @params jid
+	 */
+	ArrayList<JID> onlineJids(JID jid) throws NodeStoreException;
 
 	/**
 	 * Count node threads

src/main/java/org/buddycloud/channelserver/db/jdbc/JDBCNodeStore.java

@@ -1668,6 +1668,61 @@ public int countNodeThreads(String node) throws NodeStoreException {
 			close(selectStatement);
 		}
 	}
+
+	@Override
+	public void jidOnline(JID jid) throws NodeStoreException {
+		PreparedStatement addStatement = null;
+		try {
+			jidOffline(jid);
+			addStatement = conn.prepareStatement(dialect.addOnlineJid());
+			addStatement.setString(1, jid.toFullJID());
+			addStatement.executeUpdate();
+			addStatement.close();
+		} catch (SQLException e) {
+			throw new NodeStoreException(e);
+		} finally {
+			close(addStatement);
+		}
+	}
+	
+	@Override
+	public void jidOffline(JID jid) throws NodeStoreException {
+		PreparedStatement stmt = null;
+		try {
+			stmt = conn.prepareStatement(dialect.deleteOnlineJid());
+			stmt.setString(1, jid.toFullJID());
+			stmt.executeUpdate();
+		} catch (SQLException e) {
+			throw new NodeStoreException(e);
+		} finally {
+			close(stmt); // Will implicitly close the resultset if required
+		}
+	}
+	
+	@Override
+	public ArrayList<JID> onlineJids(JID jid) throws NodeStoreException {
+		PreparedStatement stmt = null;
+
+		try {
+			stmt = conn.prepareStatement(dialect.selectOnlineResources());
+
+			stmt.setString(1, jid.toBareJID() + "%");
+
+			java.sql.ResultSet rs = stmt.executeQuery();
+
+			ArrayList<JID> result = new ArrayList<JID>();
+
+			while (rs.next()) {
+				result.add(new JID(rs.getString(1)));
+			}
+
+			return result;
+		} catch (SQLException e) {
+			throw new NodeStoreException(e);
+		} finally {
+			close(stmt); // Will implicitly close the resultset if required
+		}
+	}
 
 	@Override
 	public Transaction beginTransaction() throws NodeStoreException {
@@ -1799,6 +1854,12 @@ private String getLocalId(String nodeItemId) {
 	public interface NodeStoreSQLDialect {
 		String insertNode();
 
+		String addOnlineJid();
+
+		String deleteOnlineJid();
+
+		String selectOnlineResources();
+
 		String selectNodeMemberships();
 
 		String selectUserMemberships();
@@ -1934,7 +1995,7 @@ public interface NodeStoreSQLDialect {
 		String selectUserFeedItems();
 
 		String selectCountUserFeedItems();
-	
+
 	}
 
 }

src/main/java/org/buddycloud/channelserver/db/jdbc/dialect/Sql92NodeStoreDialect.java

@@ -380,6 +380,15 @@ public class Sql92NodeStoreDialect implements NodeStoreSQLDialect {
 		"WHERE " +
 		    "(\"subscriptions\".\"node\" = ?) " +
 		"ORDER BY \"updated\" DESC; ";
+
+	private static final String DELETE_ONLINE_JID = "DELETE FROM \"online_users\" WHERE \"user\" = ?;";
+	
+	private static final String INSERT_ONLINE_JID = "INSERT INTO \"online_users\" (\"user\", \"updated\") VALUES (?, NOW());";
+
+	private static final String SELECT_ONLINE_RESOURCES = "SELECT \"user\", \"updated\" "
+            + "FROM \"online_users\" "
+            + "WHERE \"user\" LIKE ? "
+            + "ORDER BY \"updated\" DESC;";
 
 	private static final String SELECT_USER_FEED_ITEMS = ""
 	    + "SELECT \"node\", \"id\", \"updated\", \"xml\", \"in_reply_to\" "
@@ -733,6 +742,20 @@ public String countNodeThreads() {
 	}
 
 	@Override
+	public String deleteOnlineJid() {
+		return DELETE_ONLINE_JID;
+	}
+
+	@Override
+	public String selectOnlineResources() {
+		return SELECT_ONLINE_RESOURCES;
+	}
+
+	@Override
+	public String addOnlineJid() {
+		return INSERT_ONLINE_JID;
+	}
+	
 	public String selectUserFeedItems() {
 		return SELECT_USER_FEED_ITEMS;
 	}

src/main/java/org/buddycloud/channelserver/queue/OutQueueConsumer.java

@@ -7,6 +7,7 @@
 import org.buddycloud.channelserver.ChannelsEngine;
 import org.buddycloud.channelserver.Configuration;
 import org.buddycloud.channelserver.channel.LocalDomainChecker;
+import org.buddycloud.channelserver.db.exception.NodeStoreException;
 import org.buddycloud.channelserver.utils.users.OnlineResourceManager;
 import org.dom4j.Attribute;
 import org.xmpp.component.ComponentException;
@@ -37,7 +38,7 @@ public OutQueueConsumer(ChannelsEngine component,
 	}
 
 	@Override
-	protected void consume(Packet p) {
+	protected void consume(Packet p) throws NodeStoreException {
 
 		try {
 			if (true == isRemoteServer(p.getTo())) {

src/main/java/org/buddycloud/channelserver/queue/QueueConsumer.java

@@ -5,35 +5,39 @@
 import java.util.concurrent.Executors;
 
 import org.apache.log4j.Logger;
+import org.buddycloud.channelserver.db.exception.NodeStoreException;
 import org.xmpp.packet.Packet;
 
 public abstract class QueueConsumer {
 
-    private static final Logger LOGGER = Logger.getLogger(QueueConsumer.class);
-    
-    private ExecutorService executorService = Executors.newFixedThreadPool(5);
-    private final BlockingQueue<Packet> queue;
-    
-    public QueueConsumer(BlockingQueue<Packet> queue) {
-        this.queue = queue;
-    }
-
-    public void start() {
-        executorService.submit(new Runnable() {
-            @Override
-            public void run() {
-                while (true) {
-                    try {
-                        Packet packet = queue.take();
-                        consume(packet);
-                    } catch (InterruptedException e) {
-                        LOGGER.error(e);
-                    }
-                }
-            }
-        });
-    }
-    
-    protected abstract void consume(Packet p);
-    
+	private static final Logger LOGGER = Logger.getLogger(QueueConsumer.class);
+
+	private ExecutorService executorService = Executors.newFixedThreadPool(5);
+	private final BlockingQueue<Packet> queue;
+
+	public QueueConsumer(BlockingQueue<Packet> queue) {
+		this.queue = queue;
+	}
+
+	public void start() {
+		executorService.submit(new Runnable() {
+			@Override
+			public void run() {
+				while (true) {
+					try {
+						Packet packet = queue.take();
+						consume(packet);
+					} catch (InterruptedException e) {
+						LOGGER.error(e);
+					} catch (NodeStoreException e) {
+						// TODO Auto-generated catch block
+						LOGGER.error(e);
+					}
+				}
+			}
+		});
+	}
+
+	protected abstract void consume(Packet p) throws NodeStoreException;
+
 }