<commit_before>
package org.buddycloud.channelserver.packetprocessor.iq.namespace.pubsub.set;

import java.util.HashSet;
import java.util.Map;
import java.util.concurrent.BlockingQueue;

import org.apache.log4j.Logger;
import org.buddycloud.channelserver.channel.ChannelManager;
import org.buddycloud.channelserver.channel.Conf;
import org.buddycloud.channelserver.db.NodeStore.Transaction;
import org.buddycloud.channelserver.db.exception.NodeStoreException;
import org.buddycloud.channelserver.packetprocessor.iq.namespace.pubsub.JabberPubsub;
import org.buddycloud.channelserver.packetprocessor.iq.namespace.pubsub.PubSubElementProcessorAbstract;
import org.buddycloud.channelserver.packetprocessor.iq.namespace.pubsub.PubSubSet;
import org.buddycloud.channelserver.pubsub.accessmodel.AccessModels;
import org.buddycloud.channelserver.pubsub.affiliation.Affiliations;
import org.buddycloud.channelserver.pubsub.event.Event;
import org.buddycloud.channelserver.pubsub.model.NodeAffiliation;
import org.buddycloud.channelserver.pubsub.model.NodeSubscription;
import org.buddycloud.channelserver.pubsub.model.impl.NodeSubscriptionImpl;
import org.buddycloud.channelserver.pubsub.subscription.Subscriptions;
import org.dom4j.Document;
import org.dom4j.Element;
import org.dom4j.dom.DOMElement;
import org.xmpp.forms.DataForm;
import org.xmpp.forms.FormField;
import org.xmpp.packet.IQ;
import org.xmpp.packet.IQ.Type;
import org.xmpp.packet.JID;
import org.xmpp.packet.Message;
import org.xmpp.packet.Packet;
import org.xmpp.packet.PacketError;
import org.xmpp.resultsetmanagement.ResultSet;

public class SubscribeSet extends PubSubElementProcessorAbstract {
	private final BlockingQueue<Packet> outQueue;
	private final ChannelManager channelManager;

	private IQ request;
	private String node;
	private JID subscribingJid;

	private static final Logger logger = Logger.getLogger(SubscribeSet.class);

	public SubscribeSet(BlockingQueue<Packet> outQueue,
			ChannelManager channelManager) {
		this.outQueue = outQueue;
		this.channelManager = channelManager;
	}

	@Override
	public void process(Element elm, JID actorJID, IQ reqIQ, Element rsm)
			throws Exception {
		node = elm.attributeValue("node");
		request = reqIQ;
		if ((node == null) || (true == node.equals(""))) {
			missingNodeName();
			return;
		}
        if (false == channelManager.isLocalNode(node)) {
        	makeRemoteRequest();
        	return;
        }
		subscribingJid = request.getFrom();
		boolean isLocalNode = true;
		boolean isLocalSubscriber = false;

		if (actorJID != null) {
			subscribingJid = actorJID;
		} else {
			isLocalSubscriber = channelManager.isLocalJID(subscribingJid);
			// Check that user is registered.
			if (!isLocalSubscriber) {
				failAuthRequired();
				return;
			}
		}

		// 6.1.3.1 JIDs Do Not Match

		// Covers where we have <EMAIL>/the-balcony
		String[] jidParts = elm.attributeValue("jid").split("/");
		String jid = jidParts[0];
		if (false == subscribingJid.toBareJID().equals(jid)) {

			/*
			 * // 6.1.3.1 JIDs Do Not Match <iq type='error'
			 * from='pubsub.shakespeare.lit' to='<EMAIL>/barracks'
			 * id='sub1'> <error type='modify'> <bad-request
			 * xmlns='urn:ietf:params:xml:ns:xmpp-stanzas'/> <invalid-jid
			 * xmlns='http://jabber.org/protocol/pubsub#errors'/> </error> </iq>
			 */

			IQ reply = IQ.createResultIQ(request);
			reply.setType(Type.error);

			Element badRequest = new DOMElement("bad-request",
					new org.dom4j.Namespace("", JabberPubsub.NS_XMPP_STANZAS));
			Element nodeIdRequired = new DOMElement("invalid-jid",
					new org.dom4j.Namespace("", JabberPubsub.NS_PUBSUB_ERROR));
			Element error = new DOMElement("error");
			error.addAttribute("type", PacketError.Type.modify.toXMPP());
			error.add(badRequest);
			error.add(nodeIdRequired);
			reply.setChildElement(error);
			outQueue.put(reply);
			return;
		}

		if (false == channelManager.nodeExists(node)) {
			IQ reply = IQ.createResultIQ(request);
			reply.setType(Type.error);
			PacketError pe = new PacketError(
					PacketError.Condition.item_not_found,
					PacketError.Type.cancel);
			reply.setError(pe);
			outQueue.put(reply);
			return;
		}

		// If node is whitelist
		// 6.1.3.4 Not on Whitelist

		// Subscribe to a node.

		Transaction t = null;
		try {
			t = channelManager.beginTransaction();

			NodeSubscription nodeSubscription = channelManager
					.getUserSubscription(node, subscribingJid);
			NodeAffiliation nodeAffiliation = channelManager
					.getUserAffiliation(node, subscribingJid);

			Affiliations possibleExistingAffiliation = nodeAffiliation
					.getAffiliation();
			Subscriptions possibleExistingSubscription = nodeSubscription
					.getSubscription();

			if (Affiliations.outcast.toString().equals(
					possibleExistingAffiliation.toString())) {
				/*
				 * 6.1.3.8 Blocked <iq type='error'
				 * from='pubsub.shakespeare.lit'
				 * to='<EMAIL>.lit/barracks' id='sub1'> <error
				 * type='auth'> <forbidden
				 * xmlns='urn:ietf:params:xml:ns:xmpp-stanzas'/> </error> </iq>
				 */
				IQ reply = IQ.createResultIQ(request);
				reply.setType(Type.error);
				PacketError pe = new PacketError(
						org.xmpp.packet.PacketError.Condition.forbidden,
						org.xmpp.packet.PacketError.Type.auth);
				reply.setError(pe);
				outQueue.put(reply);
				return;
			}

			Affiliations defaultAffiliation = null;
			Subscriptions defaultSubscription = null;
			
			if (false == possibleExistingSubscription.in(Subscriptions.none)) {
				logger.debug("User already has a '"
						+ possibleExistingSubscription.toString()
						+ "' subscription");
				defaultAffiliation = possibleExistingAffiliation;
				defaultSubscription = possibleExistingSubscription;
			} else {

				// Finally subscribe to the node :-)
				Map<String, String> nodeConf = channelManager.getNodeConf(node);
				try {
					defaultAffiliation = Affiliations.createFromString(nodeConf
							.get(Conf.DEFAULT_AFFILIATION));
				} catch (NullPointerException e) {
					e.printStackTrace();
					defaultAffiliation = Affiliations.member;
				}
				defaultSubscription = Subscriptions.subscribed;
				String accessModel = nodeConf.get(Conf.ACCESS_MODEL);
				if ((null == accessModel) 
				    || (true == accessModel.equals(AccessModels.authorize.toString()))) {
					defaultSubscription = Subscriptions.pending;
				}
	
				NodeSubscription newSubscription = new NodeSubscriptionImpl(node,
						subscribingJid, request.getFrom(), defaultSubscription);
				channelManager.addUserSubscription(newSubscription);
	
				channelManager.setUserAffiliation(node, subscribingJid,
						defaultAffiliation);
			}

			IQ reply = IQ.createResultIQ(request);
			Element pubsub = reply.setChildElement(PubSubSet.ELEMENT_NAME,
					JabberPubsub.NAMESPACE_URI);
			pubsub.addElement("subscription")
					.addAttribute("node", node)
					.addAttribute("jid", subscribingJid.toBareJID())
					.addAttribute("subscription",
							defaultSubscription.toString());
			pubsub.addElement("affiliation").addAttribute("node", node)
					.addAttribute("jid", subscribingJid.toBareJID())
					.addAttribute("affiliation", defaultAffiliation.toString());

			outQueue.put(reply);

			notifySubscribers(defaultSubscription, defaultAffiliation);

			t.commit();
		} finally {
			if (t != null)
				t.close();
		}
	}

	private void makeRemoteRequest() throws InterruptedException {
		request.setTo(new JID(node.split("/")[2]).getDomain());
		Element actor = request.getElement()
		    .element("pubsub")
		    .addElement("actor", JabberPubsub.NS_BUDDYCLOUD);
		actor.addText(request.getFrom().toBareJID());
	    outQueue.put(request);
	}

	private void notifySubscribers(Subscriptions subscriptionStatus,
			Affiliations affiliationType) throws NodeStoreException,
			InterruptedException {

		ResultSet<NodeSubscription> subscribers = channelManager
				.getNodeSubscriptionListeners(node);

		// Get all the affiliated users (so we can work out moderators)
		ResultSet<NodeAffiliation> nodeAffiliations = channelManager
				.getNodeAffiliations(node);
		HashSet<JID> moderatorOwners = new HashSet<JID>();

		for (NodeAffiliation nodeAffiliation : nodeAffiliations) {
			if (nodeAffiliation.getAffiliation().in(Affiliations.owner,
					Affiliations.moderator)) {
				moderatorOwners.add(nodeAffiliation.getUser());
			}
		}<commit_msg>Support subscribing to /firehose<commit_after>
package org.buddycloud.channelserver.packetprocessor.iq.namespace.pubsub.set;

import java.util.HashSet;
import java.util.Map;
import java.util.concurrent.BlockingQueue;

import org.apache.log4j.Logger;
import org.buddycloud.channelserver.channel.ChannelManager;
import org.buddycloud.channelserver.channel.Conf;
import org.buddycloud.channelserver.db.NodeStore.Transaction;
import org.buddycloud.channelserver.db.exception.NodeStoreException;
import org.buddycloud.channelserver.packetprocessor.iq.namespace.pubsub.JabberPubsub;
import org.buddycloud.channelserver.packetprocessor.iq.namespace.pubsub.PubSubElementProcessorAbstract;
import org.buddycloud.channelserver.packetprocessor.iq.namespace.pubsub.PubSubSet;
import org.buddycloud.channelserver.pubsub.accessmodel.AccessModels;
import org.buddycloud.channelserver.pubsub.affiliation.Affiliations;
import org.buddycloud.channelserver.pubsub.event.Event;
import org.buddycloud.channelserver.pubsub.model.NodeAffiliation;
import org.buddycloud.channelserver.pubsub.model.NodeSubscription;
import org.buddycloud.channelserver.pubsub.model.impl.NodeSubscriptionImpl;
import org.buddycloud.channelserver.pubsub.subscription.Subscriptions;
import org.dom4j.Document;
import org.dom4j.Element;
import org.dom4j.dom.DOMElement;
import org.xmpp.forms.DataForm;
import org.xmpp.forms.FormField;
import org.xmpp.packet.IQ;
import org.xmpp.packet.IQ.Type;
import org.xmpp.packet.JID;
import org.xmpp.packet.Message;
import org.xmpp.packet.Packet;
import org.xmpp.packet.PacketError;
import org.xmpp.resultsetmanagement.ResultSet;

public class SubscribeSet extends PubSubElementProcessorAbstract {
	private final BlockingQueue<Packet> outQueue;
	private final ChannelManager channelManager;

	private IQ request;
	private String node;
	private JID subscribingJid;
	
	private Map<String, String> nodeConf;

	private static final Logger logger = Logger.getLogger(SubscribeSet.class);

	public SubscribeSet(BlockingQueue<Packet> outQueue,
			ChannelManager channelManager) {
		this.outQueue = outQueue;
		this.channelManager = channelManager;
	}

	@Override
	public void process(Element elm, JID actorJID, IQ reqIQ, Element rsm)
			throws Exception {
		node = elm.attributeValue("node");
		request = reqIQ;
		if ((node == null) || (true == node.equals(""))) {
			missingNodeName();
			return;
		}
		
		if (true == node.equals("/firehose")) {
			if (false == channelManager.nodeExists("/firehose"))
	            channelManager.addRemoteNode("/firehose");
			nodeConf = new HashMap<String, String>();
			nodeConf.put(Conf.DEFAULT_AFFILIATION, "member");
			nodeConf.put(Conf.ACCESS_MODEL, "open");
		} else if (false == handleNodeSubscription(elm, actorJID)) {
			return;
		}

		// Subscribe to a node.

		Transaction t = null;
		try {
			t = channelManager.beginTransaction();

			NodeSubscription nodeSubscription = channelManager
					.getUserSubscription(node, subscribingJid);
			NodeAffiliation nodeAffiliation = channelManager
					.getUserAffiliation(node, subscribingJid);

			Affiliations possibleExistingAffiliation = nodeAffiliation
					.getAffiliation();
			Subscriptions possibleExistingSubscription = nodeSubscription
					.getSubscription();

			if (Affiliations.outcast.toString().equals(
					possibleExistingAffiliation.toString())) {
				/*
				 * 6.1.3.8 Blocked <iq type='error'
				 * from='pubsub.shakespeare.lit'
				 * to='<EMAIL>/barracks' id='sub1'> <error
				 * type='auth'> <forbidden
				 * xmlns='urn:ietf:params:xml:ns:xmpp-stanzas'/> </error> </iq>
				 */
				IQ reply = IQ.createResultIQ(request);
				reply.setType(Type.error);
				PacketError pe = new PacketError(
						org.xmpp.packet.PacketError.Condition.forbidden,
						org.xmpp.packet.PacketError.Type.auth);
				reply.setError(pe);
				outQueue.put(reply);
				return;
			}

			Affiliations defaultAffiliation = null;
			Subscriptions defaultSubscription = null;
			
			if (false == possibleExistingSubscription.in(Subscriptions.none)) {
				logger.debug("User already has a '"
						+ possibleExistingSubscription.toString()
						+ "' subscription");
				defaultAffiliation = possibleExistingAffiliation;
				defaultSubscription = possibleExistingSubscription;
			} else {
				try {
					defaultAffiliation = Affiliations.createFromString(nodeConf
							.get(Conf.DEFAULT_AFFILIATION));
				} catch (NullPointerException e) {
					e.printStackTrace();
					defaultAffiliation = Affiliations.member;
				}
				defaultSubscription = Subscriptions.subscribed;
				String accessModel = nodeConf.get(Conf.ACCESS_MODEL);
				if ((null == accessModel) 
				    || (true == accessModel.equals(AccessModels.authorize.toString()))) {
					defaultSubscription = Subscriptions.pending;
				}
	
				NodeSubscription newSubscription = new NodeSubscriptionImpl(node,
						subscribingJid, request.getFrom(), defaultSubscription);
				channelManager.addUserSubscription(newSubscription);
	
				channelManager.setUserAffiliation(node, subscribingJid,
						defaultAffiliation);
			}

			IQ reply = IQ.createResultIQ(request);
			Element pubsub = reply.setChildElement(PubSubSet.ELEMENT_NAME,
					JabberPubsub.NAMESPACE_URI);
			pubsub.addElement("subscription")
					.addAttribute("node", node)
					.addAttribute("jid", subscribingJid.toBareJID())
					.addAttribute("subscription",
							defaultSubscription.toString());
			pubsub.addElement("affiliation").addAttribute("node", node)
					.addAttribute("jid", subscribingJid.toBareJID())
					.addAttribute("affiliation", defaultAffiliation.toString());

			outQueue.put(reply);

			notifySubscribers(defaultSubscription, defaultAffiliation);

			t.commit();
		} finally {
			if (t != null)
				t.close();
		}
	}

	private boolean handleNodeSubscription(Element elm, JID actorJID)
			throws NodeStoreException, InterruptedException {
		if ((false == channelManager.isLocalNode(node)) && (false == node.equals("/firehose"))) {
        	makeRemoteRequest();
        	return false;
        }
		subscribingJid = request.getFrom();
		boolean isLocalNode = true;
		boolean isLocalSubscriber = false;

		if (actorJID != null) {
			subscribingJid = actorJID;
		} else {
			isLocalSubscriber = channelManager.isLocalJID(subscribingJid);
			// Check that user is registered.
			if (!isLocalSubscriber) {
				failAuthRequired();
				return false;
			}
		}

		// 6.1.3.1 JIDs Do Not Match

		// Covers where we have <EMAIL>/the-balcony
		String[] jidParts = elm.attributeValue("jid").split("/");
		String jid = jidParts[0];
		if (false == subscribingJid.toBareJID().equals(jid)) {

			/*
			 * // 6.1.3.1 JIDs Do Not Match <iq type='error'
			 * from='pubsub.shakespeare.lit' to='<EMAIL>/barracks'
			 * id='sub1'> <error type='modify'> <bad-request
			 * xmlns='urn:ietf:params:xml:ns:xmpp-stanzas'/> <invalid-jid
			 * xmlns='http://jabber.org/protocol/pubsub#errors'/> </error> </iq>
			 */

			IQ reply = IQ.createResultIQ(request);
			reply.setType(Type.error);

			Element badRequest = new DOMElement("bad-request",
					new org.dom4j.Namespace("", JabberPubsub.NS_XMPP_STANZAS));
			Element nodeIdRequired = new DOMElement("invalid-jid",
					new org.dom4j.Namespace("", JabberPubsub.NS_PUBSUB_ERROR));
			Element error = new DOMElement("error");
			error.addAttribute("type", PacketError.Type.modify.toXMPP());
			error.add(badRequest);
			error.add(nodeIdRequired);
			reply.setChildElement(error);
			outQueue.put(reply);
			return false;
		}

		if (false == channelManager.nodeExists(node)) {
			IQ reply = IQ.createResultIQ(request);
			reply.setType(Type.error);
			PacketError pe = new PacketError(
					PacketError.Condition.item_not_found,
					PacketError.Type.cancel);
			reply.setError(pe);
			outQueue.put(reply);
			return false;
		}

		nodeConf = channelManager.getNodeConf(node);
		return true;
	}

	private void makeRemoteRequest() throws InterruptedException {
		request.setTo(new JID(node.split("/")[2]).getDomain());
		Element actor = request.getElement()
		    .element("pubsub")
		    .addElement("actor", JabberPubsub.NS_BUDDYCLOUD);
		actor.addText(request.getFrom().toBareJID());
	    outQueue.put(request);
	}

	private void notifySubscribers(Subscriptions subscriptionStatus,
			Affiliations affiliationType) throws NodeStoreException,
			InterruptedException {

		ResultSet<NodeSubscription> subscribers = channelManager
				.getNodeSubscriptionListeners(node);

		// Get all the affiliated users (so we can work out moderators)
		ResultSet<NodeAffiliation> nodeAffiliations = channelManager
				.getNodeAffiliations(node);
		HashSet<JID> moderatorOwners = new HashSet<JID>();

		for (NodeAffiliation nodeAffiliation : nodeAffiliations) {
			if (nodeAffiliation.getAffiliation().in(Affiliations.owner,
					Affiliations.moderator)) {
				moderatorOwners.add(nodeAffiliation.getUser());
			}
		}
{'fraction_non_alphanumeric': 0.11762650672567403, 'fraction_numerical': 0.002795085308332848, 'mean_word_length': 4.445149017121116, 'pattern_counts': {'":': 0, '<': 46, '<?xml version=': 0, '>': 46, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 10}, 'pii_count': 7, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3962311', 'n_tokens_mistral': 6200, 'n_tokens_neox': 5304, 'n_words': 924}