<commit_before>var io = require('socket.io-client');
var Promise = require('bluebird');
var EventEmitter = require('events').EventEmitter;
var slice = [].slice;

function createPromise(fnc){
	return new Promise(function(resolve, reject){
		fnc(function(err, result){
			if(err){
				reject(err);
			}else{
				resolve(result);
			}
		});
	});
}

function wrapSocket(socket){
	var socksub = {};
	var events = new EventEmitter();
	var subscriptions = {};
	
	socket.on('error', function(err){
		events.emit('error', err);
	});
	
	socket.on('event', function(topic, args){
		args.unshift(topic);
		events.emit.apply(events, args);
	});
	
	socket.on('reconnect', function(){
		for(var topic in subscriptions){
			socket.emit('subscribe', topic, function(err){
				if(err){
					socket.emit('error', err);
				}
			});
		}
	});
	
	socksub.subscribe = function(topic, callback){
		if(!subscriptions[topic]){
			subscriptions[topic] = createPromise(function(callback){
				socket.emit('subscribe', topic, callback);
			})['catch'](function(err){
				delete subscriptions[topic];
				throw new Error(err);
			});
		}
		return subscriptions[topic];
	};
	
	socksub.unsubscribe = function(topic, callback){
		var subscribePromise = subscriptions[topic];
		delete subscriptions[topic];
		return createPromise(function(callback){
			socket.emit('unsubscribe', topic, callback);
		})['catch'](function(err){
			subscriptions[topic] = subscribePromise;
			throw new Error(err);
		});
	};
	
	socksub.on = function(topic, handler, autoSubscribe){
		events.on(topic, handler);
		if(autoSubscribe){
			return socksub.subscribe(topic);
		}
		return socksub;
	};
	
	socksub.off = function(topic, handler, autoUnsubscribe){
		if(handler){
			events.removeListener(topic, handler);
		}else{
			events.removeAllListeners(topic);
		}
		if(autoUnsubscribe){
			if(EventEmitter.listenerCount(events, topic) === 0){
				return socksub.unsubscribe(topic);
			}else{
				return Promise.resolve();
			}
		}
		return socksub;
	};
	
	socksub.rpc = function(method){
		var args = slice.call(arguments, 1);
		return new Promise(function(resolve, reject){
			socket.emit('rpc', method, args, function(err, result){
				if(err){
					reject(err);
				}else{
					resolve(result);
				}
			});
		});
	};
	
	return socksub;
}

module.exports = function(){
	var socket = io.connect.apply(io, arguments);
	return wrapSocket(socket);
};<commit_msg>fixed error passing in RPC system<commit_after>var io = require('socket.io-client');
var Promise = require('bluebird');
var EventEmitter = require('events').EventEmitter;
var slice = [].slice;

function createPromise(fnc){
	return new Promise(function(resolve, reject){
		fnc(function(err, result){
			if(err){
				reject(new Error(err));
			}else{
				resolve(result);
			}
		});
	});
}

function wrapSocket(socket){
	var socksub = {};
	var events = new EventEmitter();
	var subscriptions = {};
	
	socket.on('error', function(err){
		events.emit('error', err);
	});
	
	socket.on('event', function(topic, args){
		args.unshift(topic);
		events.emit.apply(events, args);
	});
	
	socket.on('reconnect', function(){
		for(var topic in subscriptions){
			socket.emit('subscribe', topic, function(err){
				if(err){
					socket.emit('error', err);
				}
			});
		}
	});
	
	socksub.subscribe = function(topic, callback){
		if(!subscriptions[topic]){
			subscriptions[topic] = createPromise(function(callback){
				socket.emit('subscribe', topic, callback);
			})['catch'](function(err){
				delete subscriptions[topic];
				throw err;
			});
		}
		return subscriptions[topic];
	};
	
	socksub.unsubscribe = function(topic, callback){
		var subscribePromise = subscriptions[topic];
		delete subscriptions[topic];
		return createPromise(function(callback){
			socket.emit('unsubscribe', topic, callback);
		})['catch'](function(err){
			subscriptions[topic] = subscribePromise;
			throw err;
		});
	};
	
	socksub.on = function(topic, handler, autoSubscribe){
		events.on(topic, handler);
		if(autoSubscribe){
			return socksub.subscribe(topic);
		}
		return socksub;
	};
	
	socksub.off = function(topic, handler, autoUnsubscribe){
		if(handler){
			events.removeListener(topic, handler);
		}else{
			events.removeAllListeners(topic);
		}
		if(autoUnsubscribe){
			if(EventEmitter.listenerCount(events, topic) === 0){
				return socksub.unsubscribe(topic);
			}else{
				return Promise.resolve();
			}
		}
		return socksub;
	};
	
	socksub.rpc = function(method){
		var args = slice.call(arguments, 1);
		return createPromise(function(callback){
			socket.emit('rpc', method, args, callback);
		});
	};
	
	return socksub;
}

module.exports = function(){
	var socket = io.connect.apply(io, arguments);
	return wrapSocket(socket);
};
{'fraction_non_alphanumeric': 0.1616931216931217, 'fraction_numerical': 0.0008465608465608466, 'mean_word_length': 4.401142857142857, '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': '877451', 'n_tokens_mistral': 1740, 'n_tokens_neox': 1506, 'n_words': 301}