<commit_before>'use strict';

let net = require('net');
let Polo = require('polo');
let polo = new Polo();
let through2 = require('through2');
let _ = require('highland');
let msgpack = require('msgpack');
let lengthPrefixedStream = require('length-prefixed-stream');

function getNextService(remoteServiceName) {
	let allServices = polo.all();

	if (allServices && allServices[remoteServiceName]) {
		if (allServices[remoteServiceName].hasOwnProperty('index')) {
			if (allServices[remoteServiceName].index >= allServices[remoteServiceName].length) {
				allServices[remoteServiceName].index = 0;
			}
			return allServices[remoteServiceName][allServices[remoteServiceName].index++];
		} else {
			allServices[remoteServiceName].index = 0;
			return getNextService((remoteServiceName));
		}
	}
}

function decodeStream() {
	// TODO use consume when https://github.com/caolan/highland/issues/265 is fixed
	return through2.obj(function (chunk, enc, callback) {
		try {
			//callback(null, JSON.parse(chunk.toString()));
			callback(null, msgpack.unpack(chunk));
		} catch (err) {
			callback(err);
		}
	});
}

function encodeStream() {
	// TODO use consume when https://github.com/caolan/highland/issues/265 is fixed
	return through2.obj(function (chunk, enc, callback) {
		//callback(null, JSON.stringify(chunk));
		callback(null, msgpack.pack(chunk));
	});
}

// We are load balancing between same service type
module.exports = function (localServiceName, options) {
	let methods = {};

	options = options || {};

	function stream(args, streamingPayload) {
		// Try round robin a service
		let remoteService = getNextService(args.service);
		if (!remoteService) {
			throw new Error('Invalid service name "' + args.service + '" or service is down.');
		}

		let socket = net.connect(remoteService,
			function () {
				// Create an out stream with all the necessary transformations
				let outStream = encodeStream();
				outStream.pipe(lengthPrefixedStream.encode()).pipe(this);

				// Write the command type
				outStream.write({
					cmd: args.cmd,
					args: args.args
				});

				// Write or stream the payload
				if (streamingPayload) {
					streamingPayload.pipe(outStream);
				}
			});

		let isConnectionTerminatedCorrectly = false;

		// Return highland stream<commit_msg>cleanup<commit_after>'use strict';

let net = require('net');
let Polo = require('polo');
let polo = new Polo();
let through2 = require('through2');
let _ = require('highland');
let msgpack = require('msgpack');
let lengthPrefixedStream = require('length-prefixed-stream');

function getNextService(remoteServiceName) {
	let allServices = polo.all();

	if (allServices && allServices[remoteServiceName]) {
		if (allServices[remoteServiceName].hasOwnProperty('index')) {
			if (allServices[remoteServiceName].index >= allServices[remoteServiceName].length) {
				allServices[remoteServiceName].index = 0;
			}
			return allServices[remoteServiceName][allServices[remoteServiceName].index++];
		} else {
			allServices[remoteServiceName].index = 0;
			return getNextService((remoteServiceName));
		}
	}
}

function decodeStream() {
	// TODO use consume when https://github.com/caolan/highland/issues/265 is fixed
	return through2.obj(function (chunk, enc, callback) {
		try {
			callback(null, msgpack.unpack(chunk));
		} catch (err) {
			callback(err);
		}
	});
}

function encodeStream() {
	// TODO use consume when https://github.com/caolan/highland/issues/265 is fixed
	return through2.obj(function (chunk, enc, callback) {
		callback(null, msgpack.pack(chunk));
	});
}

// We are load balancing between same service type
module.exports = function (localServiceName, options) {
	let methods = {};

	options = options || {};

	function stream(args, streamingPayload) {

		// Try round robin a service
		let remoteService = getNextService(args.service);
		if (!remoteService) {
			throw new Error('Invalid service name "' + args.service + '" or service is down.');
		}

		let socket = net.connect(remoteService,
			function () {
				// Create an out stream with all the necessary transformations
				let outStream = encodeStream();
				outStream.pipe(lengthPrefixedStream.encode()).pipe(this);

				// Write the command type
				outStream.write({
					cmd: args.cmd,
					args: args.args
				});

				// TODO Write end if no streaming payload ?
				// Write or stream the payload
				if (streamingPayload) {
					streamingPayload.pipe(outStream);
				}
			});

		let isConnectionTerminatedCorrectly = false;

		// Return highland stream
{'fraction_non_alphanumeric': 0.13177094848551846, 'fraction_numerical': 0.005306212690692019, 'mean_word_length': 4.1940298507462686, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 4, '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': '5765248', 'n_tokens_mistral': 1501, 'n_tokens_neox': 1311, 'n_words': 330}