<commit_before>var _ = require( 'lodash' );
var AckBatch = require( '../ackBatch.js' );
var postal = require( 'postal' );
var dispatch = postal.channel( 'rabbit.dispatch' );
var responses = postal.channel( 'rabbit.responses' );
var when = require( 'when' );
var log = require( '../log.js' )( 'wascally.amqp-queue' );
var topLog = require( '../log.js' )( 'wascally.topology' );
var unhandledLog = require( '../log.js' )( 'wascally.unhandled' );
var noOp = function() {};

function aliasOptions( options, aliases ) {
	var aliased = _.transform( options, function( result, value, key ) {
		var alias = aliases[ key ];
		result[ alias || key ] = value;
	} );
	return _.omit( aliased, Array.prototype.slice.call( arguments, 2 ) );
}

function define( channel, options, subscriber, connectionName ) {
	var valid = aliasOptions( options, {
		queuelimit: 'maxLength',
		queueLimit: 'maxLength',
		deadletter: 'deadLetterExchange',
		deadLetter: 'deadLetterExchange',
		deadLetterRoutingKey: 'deadLetterRoutingKey'
	}, 'subscribe', 'limit', 'noBatch' );
	topLog.info( 'Declaring queue \'%s\' on connection \'%s\' with the options: %s',
		options.name, connectionName, JSON.stringify( _.omit( options, [ 'name' ] ) ) );
	return channel.assertQueue( options.name, valid )
		.then( function( q ) {
			if ( options.limit ) {
				channel.prefetch( options.limit );
			}
			if ( options.subscribe ) {
				subscriber();
			}
			return q;
		} );
}

function destroy( channel, options, messages, released ) {
	function finalize() {
		messages.ignoreSignal();
		channel.destroy();
		channel = undefined;
	}

	function onUnsubscribed() {
		return when.promise( function( resolve ) {
			if ( messages.messages.length && !released ) {
				messages.once( 'empty', function() {
					finalize();
					resolve();
				} );
			} else {
				finalize();
				resolve();
			}
		} );
	}

	return unsubscribe( channel, options )
		.then( onUnsubscribed, onUnsubscribed );
}

function getChannel( connection ) {
	return connection.createChannel( true );
}

function getCount( messages ) {
	if ( messages ) {
		return messages.messages.length;
	} else {
		return 0;
	}
}

function getReply( channel, raw, replyQueue, connectionName ) {
	var position = 0;
	return function( reply, more, replyType ) {
		if ( _.isString( more ) ) {
			replyType = more;
			more = false;
		}
		var replyTo = raw.properties.replyTo;
		raw.ack();
		if ( replyTo ) {
			var payload = new Buffer( JSON.stringify( reply ) ),
				publishOptions = {
					type: replyType || raw.type + '.reply',
					contentType: 'application/json',
					contentEncoding: 'utf8',
					correlationId: raw.properties.messageId,
					replyTo: replyQueue === false ? undefined : replyQueue,
					headers: {}
				};
			if ( !more ) {
				publishOptions.headers.sequence_end = true; // jshint ignore:line
			} else {
				publishOptions.headers.position = ( position++ );
			}
			log.debug( 'Replying to message %s on %s - %s with type %s',
				raw.properties.messageId,
				replyTo,
				connectionName,
				publishOptions.type );
			if ( raw.properties.headers[ 'direct-reply-to' ] ) {
				return channel.publish(
					'',
					replyTo,
					payload,<commit_msg>Added check if headers existed on reply<commit_after>var _ = require( 'lodash' );
var AckBatch = require( '../ackBatch.js' );
var postal = require( 'postal' );
var dispatch = postal.channel( 'rabbit.dispatch' );
var responses = postal.channel( 'rabbit.responses' );
var when = require( 'when' );
var log = require( '../log.js' )( 'wascally.amqp-queue' );
var topLog = require( '../log.js' )( 'wascally.topology' );
var unhandledLog = require( '../log.js' )( 'wascally.unhandled' );
var noOp = function() {};

function aliasOptions( options, aliases ) {
	var aliased = _.transform( options, function( result, value, key ) {
		var alias = aliases[ key ];
		result[ alias || key ] = value;
	} );
	return _.omit( aliased, Array.prototype.slice.call( arguments, 2 ) );
}

function define( channel, options, subscriber, connectionName ) {
	var valid = aliasOptions( options, {
		queuelimit: 'maxLength',
		queueLimit: 'maxLength',
		deadletter: 'deadLetterExchange',
		deadLetter: 'deadLetterExchange',
		deadLetterRoutingKey: 'deadLetterRoutingKey'
	}, 'subscribe', 'limit', 'noBatch' );
	topLog.info( 'Declaring queue \'%s\' on connection \'%s\' with the options: %s',
		options.name, connectionName, JSON.stringify( _.omit( options, [ 'name' ] ) ) );
	return channel.assertQueue( options.name, valid )
		.then( function( q ) {
			if ( options.limit ) {
				channel.prefetch( options.limit );
			}
			if ( options.subscribe ) {
				subscriber();
			}
			return q;
		} );
}

function destroy( channel, options, messages, released ) {
	function finalize() {
		messages.ignoreSignal();
		channel.destroy();
		channel = undefined;
	}

	function onUnsubscribed() {
		return when.promise( function( resolve ) {
			if ( messages.messages.length && !released ) {
				messages.once( 'empty', function() {
					finalize();
					resolve();
				} );
			} else {
				finalize();
				resolve();
			}
		} );
	}

	return unsubscribe( channel, options )
		.then( onUnsubscribed, onUnsubscribed );
}

function getChannel( connection ) {
	return connection.createChannel( true );
}

function getCount( messages ) {
	if ( messages ) {
		return messages.messages.length;
	} else {
		return 0;
	}
}

function getReply( channel, raw, replyQueue, connectionName ) {
	var position = 0;
	return function( reply, more, replyType ) {
		if ( _.isString( more ) ) {
			replyType = more;
			more = false;
		}
		var replyTo = raw.properties.replyTo;
		raw.ack();
		if ( replyTo ) {
			var payload = new Buffer( JSON.stringify( reply ) ),
				publishOptions = {
					type: replyType || raw.type + '.reply',
					contentType: 'application/json',
					contentEncoding: 'utf8',
					correlationId: raw.properties.messageId,
					replyTo: replyQueue === false ? undefined : replyQueue,
					headers: {}
				};
			if ( !more ) {
				publishOptions.headers.sequence_end = true; // jshint ignore:line
			} else {
				publishOptions.headers.position = ( position++ );
			}
			log.debug( 'Replying to message %s on %s - %s with type %s',
				raw.properties.messageId,
				replyTo,
				connectionName,
				publishOptions.type );
			if ( raw.properties.headers && raw.properties.headers[ 'direct-reply-to' ] ) {
				return channel.publish(
					'',
					replyTo,
					payload,
{'fraction_non_alphanumeric': 0.14959400374765772, 'fraction_numerical': 0.0012492192379762648, 'mean_word_length': 3.426399447131997, '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': '1500870', 'n_tokens_mistral': 2355, 'n_tokens_neox': 2020, 'n_words': 518}