<commit_before>package org.buddycloud.channelserver.queue;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;
import org.buddycloud.channelserver.ChannelsEngine;
import org.buddycloud.channelserver.packetprocessor.iq.namespace.pubsub.JabberPubsub;
import org.dom4j.Attribute;
import org.dom4j.Element;
import org.dom4j.Namespace;
import org.dom4j.dom.DOMElement;
import org.hsqldb.Server;
import org.xmpp.component.ComponentException;
import org.xmpp.packet.IQ;
import org.xmpp.packet.JID;
import org.xmpp.packet.Packet;
import org.xmpp.packet.PacketError;

public class FederatedQueueManager {
	private static final Logger logger = Logger
			.getLogger(FederatedQueueManager.class);

	public static final String NO_CHANNEL_SERVER = "NO_CHANNEL_SERVER";
	public static final String DISCO_ITEMS = "DISCO_ITEMS";
	public static final String DISCO_INFO = "DISCO_INFO";
	public static final String DISCOVERED = "DISCOVERED";

	public static final String IDENTITY_TYPE_CHANNELS = "channels";
	public static final String BUDDYCLOUD_SERVER = "buddycloud-server";

	private int id = 1;

	private final ChannelsEngine component;
	private HashMap<String, String> discoveredServers = new HashMap<String, String>();
	private HashMap<String, List<Packet>> waitingStanzas = new HashMap<String, List<Packet>>();
	private HashMap<String, String> remoteChannelDiscoveryStatus = new HashMap<String, String>();
	private HashMap<String, Integer> remoteServerItemsToProcess = new HashMap<String, Integer>();
	private HashMap<String, String> remoteServerInfoRequestIds = new HashMap<String, String>();
	private HashMap<String, JID> sentRemotePackets = new HashMap<String, JID>();
	private HashMap<String, String> nodeMap = new HashMap<String, String>();

	private String localServer;

	private BlockingQueue<Packet> federatedResponseQueue;

	public FederatedQueueManager(ChannelsEngine component, String localServer) {
		this.component = component;
		this.localServer = localServer;
	}

	private int getId() {
		int id = this.id;
		this.id++;
		return id;
	}

	public void process(Packet packet) throws ComponentException {
		String to = packet.getTo().toString();
		sentRemotePackets.put(packet.getID(), packet.getFrom());
		packet.setFrom(localServer);
		try {
			extractNodeDetails(packet);
			// Do we have a map already?
			if (discoveredServers.containsKey(to)) {
				packet.setTo(new JID(discoveredServers.get(to)));
				sendPacket(packet.createCopy());
				return;
			}
			// Are we already discovering a remote server?
			if (false == remoteChannelDiscoveryStatus.containsKey(to)) {
				discoverRemoteChannelServer(to, packet.getID());
			} else if (remoteChannelDiscoveryStatus.get(to).equals(
					NO_CHANNEL_SERVER)) {
				logger.error("No remote channel server for " + to);
				IQ reply = IQ.createResultIQ((IQ) packet);
				reply.setError(new PacketError(
						PacketError.Condition.remote_server_not_found,
						PacketError.Type.cancel));
				component.sendPacket(reply);
				return;
			}
			// Add packet to list
			if (false == waitingStanzas.containsKey(to)) {
				waitingStanzas.put(to, new ArrayList<Packet>());
			}
			logger.debug("Adding packet to waiting stanza list for " + to
					+ " (size " + waitingStanzas.get(to).size() + ")");
			waitingStanzas.get(to).add(packet);
		} catch (Exception e) {
			logger.error(e);
		}
	}

	private void extractNodeDetails(Packet packet) {
		try {
			String packetXml = packet.toXML();
			if (false == packetXml.contains("node=")) return;
			nodeMap.put(
			    packet.getID(),
			    packetXml.split("node=\"")[1].split("\"")[0]
			);
		} catch (NullPointerException e) {
			logger.info("No node details found in federated packet");
			logger.error(e);
		} catch (ArrayIndexOutOfBoundsException e) {
			logger.info("Error extracting node information from federated packet");
			logger.error(e);
		}
	}

	private void sendPacket(Packet packet) throws ComponentException {
		logger.info("OUT-> " + packet.toXML());
		component.sendPacket(packet.createCopy());
	}

	private void discoverRemoteChannelServer(String remoteDomain, String id)
			throws ComponentException {
		logger.info("Attemping to discover remote server " + remoteDomain);
		IQ discover = new IQ(IQ.Type.get);
		discover.setFrom(localServer);
		discover.setTo(remoteDomain);
		discover.setID(id);
		discover.getElement().addElement("query", JabberPubsub.NS_DISCO_ITEMS);
		component.sendPacket(discover);
		remoteChannelDiscoveryStatus.put(remoteDomain, DISCO_ITEMS);
	}

	public void sendInfoRequests(JID from, List<Element> items)
			throws ComponentException {

		for (Element item : items) {
			Attribute name = item.attribute("name");
			if ((null != name)
					&& (true == name.getStringValue().equals(BUDDYCLOUD_SERVER))) {
				remoteChannelDiscoveryStatus.put(from.toString(), DISCOVERED);
				setDiscoveredServer(from.toString(), item.attributeValue("jid"));
				sendFederatedRequests(from.toString());
				return;
			}
		}
		IQ infoRequest = new IQ(IQ.Type.get);
		infoRequest.setFrom(localServer);
		infoRequest.getElement()
				.addElement("query", JabberPubsub.NS_DISCO_INFO);

		remoteServerItemsToProcess.put(from.toString(), items.size());
		String infoRequestId;
		for (Element item : items) {
			infoRequestId = getId() + ":info";
			infoRequest.setTo(item.attributeValue("jid"));
			infoRequest.setID(infoRequestId);
			remoteServerInfoRequestIds.put(infoRequestId, from.toString());
			component.sendPacket(infoRequest.createCopy());
		}
		remoteChannelDiscoveryStatus.put(from.toString(), DISCO_INFO);
	}

	private void setDiscoveredServer(String server, String handler) {
		discoveredServers.put(server, handler);
	}

	public void processInfoResponses(JID from, String id,
			List<Element> identities) throws ComponentException {
		String originatingServer = remoteServerInfoRequestIds.get(id);
		remoteServerInfoRequestIds.remove(id);
		remoteServerItemsToProcess.put(originatingServer,
				remoteServerItemsToProcess.get(originatingServer) - 1);

		String identityType;
		for (Element identity : identities) {
			identityType = identity.attributeValue("type");
			if ((identityType != null)
					&& (true == identityType.equals(IDENTITY_TYPE_CHANNELS))) {
				setDiscoveredServer(originatingServer, from.toString());
				sendFederatedRequests(originatingServer);
			}
		}
		if (remoteServerItemsToProcess.get(originatingServer) < 1) {
			remoteServerItemsToProcess.remove(originatingServer);
			if (false == discoveredServers.containsKey(originatingServer)) {
				sendRemoteChannelServerNotFoundErrorResponses(originatingServer);
				remoteChannelDiscoveryStatus.put(originatingServer,
						NO_CHANNEL_SERVER);
				waitingStanzas.remove(originatingServer);
			} else {
				remoteChannelDiscoveryStatus.put(originatingServer, DISCOVERED);
			}
		}
	}

	private void sendFederatedRequests(String originatingServer)
			throws ComponentException {
		String remoteServer = discoveredServers.get(originatingServer);
		List<Packet> packetsToSend = waitingStanzas.get(originatingServer);
		if (null == packetsToSend) {
			return;
		}
		for (Packet packet : packetsToSend) {
			packet.setTo(remoteServer);
			sendPacket(packet);
		}<commit_msg>Do not remove index as later packets will cause exceptions<commit_after>package org.buddycloud.channelserver.queue;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;
import org.buddycloud.channelserver.ChannelsEngine;
import org.buddycloud.channelserver.packetprocessor.iq.namespace.pubsub.JabberPubsub;
import org.dom4j.Attribute;
import org.dom4j.Element;
import org.dom4j.Namespace;
import org.dom4j.dom.DOMElement;
import org.hsqldb.Server;
import org.xmpp.component.ComponentException;
import org.xmpp.packet.IQ;
import org.xmpp.packet.JID;
import org.xmpp.packet.Packet;
import org.xmpp.packet.PacketError;

public class FederatedQueueManager {
	private static final Logger logger = Logger
			.getLogger(FederatedQueueManager.class);

	public static final String NO_CHANNEL_SERVER = "NO_CHANNEL_SERVER";
	public static final String DISCO_ITEMS = "DISCO_ITEMS";
	public static final String DISCO_INFO = "DISCO_INFO";
	public static final String DISCOVERED = "DISCOVERED";

	public static final String IDENTITY_TYPE_CHANNELS = "channels";
	public static final String BUDDYCLOUD_SERVER = "buddycloud-server";

	private int id = 1;

	private final ChannelsEngine component;
	private HashMap<String, String> discoveredServers = new HashMap<String, String>();
	private HashMap<String, List<Packet>> waitingStanzas = new HashMap<String, List<Packet>>();
	private HashMap<String, String> remoteChannelDiscoveryStatus = new HashMap<String, String>();
	private HashMap<String, Integer> remoteServerItemsToProcess = new HashMap<String, Integer>();
	private HashMap<String, String> remoteServerInfoRequestIds = new HashMap<String, String>();
	private HashMap<String, JID> sentRemotePackets = new HashMap<String, JID>();
	private HashMap<String, String> nodeMap = new HashMap<String, String>();

	private String localServer;

	private BlockingQueue<Packet> federatedResponseQueue;

	public FederatedQueueManager(ChannelsEngine component, String localServer) {
		this.component = component;
		this.localServer = localServer;
	}

	private int getId() {
		int id = this.id;
		this.id++;
		return id;
	}

	public void process(Packet packet) throws ComponentException {
		String to = packet.getTo().toString();
		sentRemotePackets.put(packet.getID(), packet.getFrom());
		packet.setFrom(localServer);
		try {
			extractNodeDetails(packet);
			// Do we have a map already?
			if (discoveredServers.containsKey(to)) {
				packet.setTo(new JID(discoveredServers.get(to)));
				sendPacket(packet.createCopy());
				return;
			}
			// Are we already discovering a remote server?
			if (false == remoteChannelDiscoveryStatus.containsKey(to)) {
				discoverRemoteChannelServer(to, packet.getID());
			} else if (remoteChannelDiscoveryStatus.get(to).equals(
					NO_CHANNEL_SERVER)) {
				logger.error("No remote channel server for " + to);
				IQ reply = IQ.createResultIQ((IQ) packet);
				reply.setError(new PacketError(
						PacketError.Condition.remote_server_not_found,
						PacketError.Type.cancel));
				component.sendPacket(reply);
				return;
			}
			// Add packet to list
			if (false == waitingStanzas.containsKey(to)) {
				waitingStanzas.put(to, new ArrayList<Packet>());
			}
			logger.debug("Adding packet to waiting stanza list for " + to
					+ " (size " + waitingStanzas.get(to).size() + ")");
			waitingStanzas.get(to).add(packet);
		} catch (Exception e) {
			logger.error(e);
		}
	}

	private void extractNodeDetails(Packet packet) {
		try {
			String packetXml = packet.toXML();
			if (false == packetXml.contains("node=")) return;
			nodeMap.put(
			    packet.getID(),
			    packetXml.split("node=\"")[1].split("\"")[0]
			);
		} catch (NullPointerException e) {
			logger.info("No node details found in federated packet");
			logger.error(e);
		} catch (ArrayIndexOutOfBoundsException e) {
			logger.info("Error extracting node information from federated packet");
			logger.error(e);
		}
	}

	private void sendPacket(Packet packet) throws ComponentException {
		logger.info("OUT-> " + packet.toXML());
		component.sendPacket(packet.createCopy());
	}

	private void discoverRemoteChannelServer(String remoteDomain, String id)
			throws ComponentException {
		logger.info("Attemping to discover remote server " + remoteDomain);
		IQ discover = new IQ(IQ.Type.get);
		discover.setFrom(localServer);
		discover.setTo(remoteDomain);
		discover.setID(id);
		discover.getElement().addElement("query", JabberPubsub.NS_DISCO_ITEMS);
		component.sendPacket(discover);
		remoteChannelDiscoveryStatus.put(remoteDomain, DISCO_ITEMS);
	}

	public void sendInfoRequests(JID from, List<Element> items)
			throws ComponentException {

		for (Element item : items) {
			Attribute name = item.attribute("name");
			if ((null != name)
					&& (true == name.getStringValue().equals(BUDDYCLOUD_SERVER))) {
				remoteChannelDiscoveryStatus.put(from.toString(), DISCOVERED);
				setDiscoveredServer(from.toString(), item.attributeValue("jid"));
				sendFederatedRequests(from.toString());
				return;
			}
		}
		IQ infoRequest = new IQ(IQ.Type.get);
		infoRequest.setFrom(localServer);
		infoRequest.getElement()
				.addElement("query", JabberPubsub.NS_DISCO_INFO);

		remoteServerItemsToProcess.put(from.toString(), items.size());
		String infoRequestId;
		for (Element item : items) {
			infoRequestId = getId() + ":info";
			infoRequest.setTo(item.attributeValue("jid"));
			infoRequest.setID(infoRequestId);
			remoteServerInfoRequestIds.put(infoRequestId, from.toString());
			component.sendPacket(infoRequest.createCopy());
		}
		remoteChannelDiscoveryStatus.put(from.toString(), DISCO_INFO);
	}

	private void setDiscoveredServer(String server, String handler) {
		discoveredServers.put(server, handler);
	}

	public void processInfoResponses(JID from, String id,
			List<Element> identities) throws ComponentException {
		String originatingServer = remoteServerInfoRequestIds.get(id);
		remoteServerInfoRequestIds.remove(id);
		remoteServerItemsToProcess.put(originatingServer,
				remoteServerItemsToProcess.get(originatingServer) - 1);

		String identityType;
		for (Element identity : identities) {
			identityType = identity.attributeValue("type");
			if ((identityType != null)
					&& (true == identityType.equals(IDENTITY_TYPE_CHANNELS))) {
				setDiscoveredServer(originatingServer, from.toString());
				sendFederatedRequests(originatingServer);
			}
		}
		if (remoteServerItemsToProcess.get(originatingServer) < 1) {
			if (false == discoveredServers.containsKey(originatingServer)) {
				sendRemoteChannelServerNotFoundErrorResponses(originatingServer);
				remoteChannelDiscoveryStatus.put(originatingServer,
						NO_CHANNEL_SERVER);
				waitingStanzas.remove(originatingServer);
			} else {
				remoteChannelDiscoveryStatus.put(originatingServer, DISCOVERED);
			}
		}
	}

	private void sendFederatedRequests(String originatingServer)
			throws ComponentException {
		String remoteServer = discoveredServers.get(originatingServer);
		List<Packet> packetsToSend = waitingStanzas.get(originatingServer);
		if (null == packetsToSend) {
			return;
		}
		for (Packet packet : packetsToSend) {
			packet.setTo(remoteServer);
			sendPacket(packet);
		}
{'fraction_non_alphanumeric': 0.11376999863257213, 'fraction_numerical': 0.0013674278681799535, 'mean_word_length': 5.2163195920101995, 'pattern_counts': {'":': 2, '<': 47, '<?xml version=': 0, '>': 47, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, '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': '3112113', 'n_tokens_mistral': 4904, 'n_tokens_neox': 4340, 'n_words': 906}