<commit_before>var url = require('url'),
	adapters = require('./adapters'),
	streams = adapters.streams,
	processors = require('./processors').processors,
	cache = {},
	aliases = {};
	


module.exports = function(){
	return function(req, res, next){
		//Get the requested URL, less the starting '/'
		var requested = url.parse(req.url).pathname.substring(1);
		
		if(requested.length == 0){
			next();
		}
		
		if(cache[requested]){
			var entry = cache[requested];
			if(entry.headers){
				for(var i in entry.headers){
					res.setHeader(i, entry.headers[i]);
				}
			}
			res.end(entry.val);
			return;
		}
		
		var segments = resolve(requested.split('.'));
		var stream = processSegments(segments,
			//onSuccess
			function(stream, context){
				
				if(context.headers){
					for(var i in context.headers){
						res.setHeader(i, context.headers[i]);
					}
				}
				
				if(context.cacheable === true){
					
					cache[requested] = {
						val: new Buffer(0)
					};
					
					cache[requested].headers = context.headers;
					
					stream.on('data', function(chunk){
						if(Buffer.isBuffer(chunk)){
							cache[requested].val = Buffer.concat([cache[requested].val, chunk]);
						} else {
							cache[requested].val = Buffer.concat([cache[requested].val, new Buffer(chunk)]);
						}
						
					});
				}
				stream.on('data', function(chunk){
					res.write(chunk);
				});
				
				stream.on('end', function(){
					res.end();
				})
				
				stream.resume();
			//	stream.pipe(res);
			},
			//onError
			function(error){
				next(error);
			},
			function onNotFound(){
				next();
			}
		);
	};
}

module.exports.stream = adapters.registerStream;
module.exports.alias = function(name, alias){
	if(streams[name] || processors[name]){
		throw name+" cannot be used as an alias: it is already a stream or a processor.";
	}
	aliases[name]=alias;
}

function resolve(segments){
	var result = [];
	for(var i=0; i<segments.length; i++){
		var seg = segments[i];
		if(aliases[seg]){
			result = result.concat(resolve(aliases[seg].split('.')));
		} else {
			result.push(seg);
		}
	}
	return result;
}

function processSegments(segments, onSuccess, onError, onNotFound){
	
	var params = {
		resultStream: null
	}
	
	var nexts = [];
	var context = {
		headers: {},
		cacheable: false
	}
	//Construct event chain look ahead.
	for(var i=0; i<segments.length; i++){
		nexts[i]= getProcessSegment(segments[i], i+1, onSuccess, onError, onNotFound, params, nexts, context);
	}
	//Build the chain
	nexts[0]();
}

function getProcessSegment(seg, nextIndex, onSuccess, onError, onNotFound, params, nexts, context){
	return function(){
		var next;
		if(nextIndex>nexts.length-1){
			next = function(){ 
						onSuccess(params.resultStream, context);
					};
		} else {
			next = nexts[nextIndex];
		}

		if(streams[seg]){
			streams[seg].init( function(stream){
				if(params.resultStream){
					processors.cat.init(function(st){
						if(!st){
							onError("Cat returned unll stream");
							return;
						}
						params.resultStream = st;
						next();
						}, context, params.resultStream, stream);
					
				} else {
					params.resultStream = stream;
					if(!params.resultStream){
						onError(seg+" returned null stream");
						return;
					}
				}
				next()
			}, context);	
		} else if(processors[seg]){<commit_msg>Bug fix<commit_after>var url = require('url'),
	adapters = require('./adapters'),
	streams = adapters.streams,
	processors = require('./processors').processors,
	cache = {},
	aliases = {};
	


module.exports = function(){
	return function(req, res, next){
		//Get the requested URL, less the starting '/'
		var requested = url.parse(req.url).pathname.substring(1);
		
		if(requested.length == 0){
			next();
		}
		
		if(cache[requested]){
			var entry = cache[requested];
			if(entry.headers){
				for(var i in entry.headers){
					res.setHeader(i, entry.headers[i]);
				}
			}
			res.end(entry.val);
			return;
		}
		
		var segments = resolve(requested.split('.'));
		var stream = processSegments(segments,
			//onSuccess
			function(stream, context){
				
				if(context.headers){
					for(var i in context.headers){
						res.setHeader(i, context.headers[i]);
					}
				}
				
				if(context.cacheable === true){
					
					cache[requested] = {
						val: new Buffer(0)
					};
					
					cache[requested].headers = context.headers;
					
					stream.on('data', function(chunk){
						if(Buffer.isBuffer(chunk)){
							cache[requested].val = Buffer.concat([cache[requested].val, chunk]);
						} else {
							cache[requested].val = Buffer.concat([cache[requested].val, new Buffer(chunk)]);
						}
						
					});
				}
				stream.on('data', function(chunk){
					res.write(chunk);
				});
				
				stream.on('end', function(){
					res.end();
				})
				
				stream.resume();
			//	stream.pipe(res);
			},
			//onError
			function(error){
				next(error);
			},
			function onNotFound(){
				next();
				return;
			}
		);
	};
}

module.exports.stream = adapters.registerStream;
module.exports.alias = function(name, alias){
	if(streams[name] || processors[name]){
		throw name+" cannot be used as an alias: it is already a stream or a processor.";
	}
	aliases[name]=alias;
}

function resolve(segments){
	var result = [];
	for(var i=0; i<segments.length; i++){
		var seg = segments[i];
		if(aliases[seg]){
			result = result.concat(resolve(aliases[seg].split('.')));
		} else {
			result.push(seg);
		}
	}
	return result;
}

function processSegments(segments, onSuccess, onError, onNotFound){
	
	var params = {
		resultStream: null
	}
	
	var nexts = [];
	var context = {
		headers: {},
		cacheable: false
	}
	//Construct event chain look ahead.
	for(var i=0; i<segments.length; i++){
		nexts[i]= getProcessSegment(segments[i], i+1, onSuccess, onError, onNotFound, params, nexts, context);
	}
	//Build the chain
	nexts[0]();
}

function getProcessSegment(seg, nextIndex, onSuccess, onError, onNotFound, params, nexts, context){
	return function(){
		var next;
		if(nextIndex>nexts.length-1){
			next = function(){ 
						onSuccess(params.resultStream, context);
					};
		} else {
			next = nexts[nextIndex];
		}
		if(streams[seg]){
			streams[seg].init( function(stream){
				if(params.resultStream){
					processors.cat.init(function(st){
						if(!st){
							onError("Cat returned null stream");
							return;
						}
						params.resultStream = st;
						next();
						}, context, params.resultStream, stream);
					
				} else {
					params.resultStream = stream;
					if(!params.resultStream){
						onError(seg+" returned null stream");
						return;
					}
					next();
				}
				
			}, context);	
		} else if(processors[seg]){
{'fraction_non_alphanumeric': 0.16177351707609347, 'fraction_numerical': 0.002396644697423607, 'mean_word_length': 2.932273262661955, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 5, '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': '2172838', 'n_tokens_mistral': 2866, 'n_tokens_neox': 2199, 'n_words': 449}