<commit_before>
var amqp = require('amqp');
var uuid = require('node-uuid');
var url = require('url');

var amqpurl = "amqp://localhost";

var implOpts = {
    reconnect: true,
    reconnectBackoffStrategy: 'linear',
    reconnectBackoffTime: 500 // ms
};

module.connection = amqp.createConnection({ url: amqpurl }, implOpts);
module.connection.on('ready', function() {
    module.connection.exchange("muon-resource", {
        durable:false,
        autoDelete:false,
        confirm: true
    }, function(exch) {

        module.resourceExchange = exch;
    });
    module.connection.exchange("muon-broadcast", {
        durable:false,
        autoDelete:false
    }, function(exch) {
        module.broadcastExchange = exch;

        startAnnouncements();
    });
});
module.connection.on('error', function (msg) {
    console.log("Getting an error");
    console.dir(msg);
});

module.discoveredServiceList = [];
module.discoveredServices = [];

setTimeout(function() {
   //todo, bootup the announcement/ detection
});

module.exports.setServiceIdentifier = function(serviceIdentifier) {
    module.serviceIdentifier = serviceIdentifier;
	console.log("Setting service identifier.");
}

module.exports.emit= function(event) {

    var headers = {};
    if (event.headers instanceof Object) {
        headers = event.headers;
    }

    var options = {
        headers:headers
    };

    var exch = module.broadcastExchange;
    exch.publish(
        event.name,
        JSON.stringify(event.payload), options, function(resp) {
    });
};

module.exports.sendAndWaitForReply = function (event, callback) {
	
	//get the url elements
	
	var u = url.parse(event.url, true);
	
	var locImplOpts = {
		reconnect: true,
		reconnectBackoffStrategy: 'linear',
		reconnectBackoffTime: 2000 // ms
	};
	
	var queue = u.hostname + "." + u.path + "." + event.method;
	
	connection = amqp.createConnection({ url: amqpurl }, locImplOpts);
	
	connection.on('ready', function() {
		
		console.log("Connected");
		
		connection.exchange("muon-resource", {
			durable:false,
			autoDelete:false,
			confirm: true
		}, function(exch) {
			
			var options = {
				replyTo : queue,
				contentType: "text/plain"
			};
			
			
			console.log('exchange set');
			connection.queue(queue, {
				durable: false,
				exclusive: true,
				ack: true,
				autoDelete: true
			}, function (q) {
				
				console.log("Creating queue " + queue);
				exch.publish(queue, event.payload, options, function(test) {
					console.log("publishing to queue");
					if(test) {
						console.log('There was an error I think');
					}
					
					callback(test);
					// 				
				});
				
			});
			
		});
		
	});
	
};<commit_msg>Fixed up sendAndWaitForReply to actually wait for the reply.<commit_after>
var _this = module.exports;

var amqp = require('amqp');
var uuid = require('node-uuid');
var url = require('url');

var amqpurl = "amqp://localhost";

var implOpts = {
    reconnect: true,
    reconnectBackoffStrategy: 'linear',
    reconnectBackoffTime: 500 // ms
};

module.connection = amqp.createConnection({ url: amqpurl }, implOpts);
module.connection.on('ready', function() {
    module.connection.exchange("muon-resource", {
        durable:false,
        autoDelete:false,
        confirm: true
    }, function(exch) {

        module.resourceExchange = exch;
    });
    module.connection.exchange("muon-broadcast", {
        durable:false,
        autoDelete:false
    }, function(exch) {
        module.broadcastExchange = exch;

        startAnnouncements();
    });
});
module.connection.on('error', function (msg) {
    console.log("Getting an error");
    console.dir(msg);
});

module.discoveredServiceList = [];
module.discoveredServices = [];

setTimeout(function() {
   //todo, bootup the announcement/ detection
});

module.exports.setServiceIdentifier = function(serviceIdentifier) {
    module.serviceIdentifier = serviceIdentifier;
	console.log("Setting service identifier.");
}

module.exports.emit= function(event) {

    var headers = {};
    if (event.headers instanceof Object) {
        headers = event.headers;
    }

    var options = {
        headers:headers
    };

    var exch = module.broadcastExchange;
    exch.publish(
        event.name,
        JSON.stringify(event.payload), options, function(resp) {
    });
};

module.exports.sendAndWaitForReply = function (event, callback) {
	
	//get the url elements
	
	var u = url.parse(event.url, true);
	
	var locImplOpts = {
		reconnect: true,
		reconnectBackoffStrategy: 'linear',
		reconnectBackoffTime: 2000 // ms
	};
	
	var queue = u.hostname + "." + u.path + "." + event.method;
	var replyQueue = queue + ".reply";
	
	connection = amqp.createConnection({ url: amqpurl }, locImplOpts);
	
	connection.on('ready', function() {
		
		console.log("Connected");
		
		connection.exchange("muon-resource", {
			durable:false,
			autoDelete:false,
			confirm: true
		}, function(exch) {
			
			var options = {
				replyTo : replyQueue,
				contentType: "text/plain"
			};

			console.log('exchange set');
			
			connection.queue(replyQueue, {
				durable: false,
				exclusive: true,
				ack: true,
				autoDelete: true
			}, function (q) {
				
				console.log("Creating queue " + replyQueue);
				
				q.bind("muon-resource", replyQueue, function () {
					console.log("Bound resource queue " + replyQueue);
					
					exch.publish(queue, event.payload, options, function(test) {
						console.log("publishing to queue");
						if(test) {
							console.log('There was an error I think');
							callback({failure: true});
						}
					});
					q.subscribe(function (message, headers, deliveryInfo, messageObject) {
						//var replyTo = messageObject.replyTo;
						console.log("Got a response");
						console.dir(message);
						callback(message);
						
						
					});
				});
				
				
				
				
			});
			
		});
		
	});
	
};
{'fraction_non_alphanumeric': 0.14086423953725757, 'fraction_numerical': 0.0023817625042531474, 'mean_word_length': 2.906312292358804, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '4524569', 'n_tokens_mistral': 2110, 'n_tokens_neox': 1755, 'n_words': 364}