schedules( data ) {
		debug('listener schedules')
		
		var _this = this;
		
		// is cache false
		if(isFalse(data.cache)) {
			this.epg.onTalk('schedules', (results) => {
				//this.emit(['schedules', data.iden], results);
				//if(data._broadcast) {
				//	this.broadcast(['schedules', data.iden], results);
				//} 
			});
			return this.epg.agent.schedules(data);
		}
		
		if(!data.lineup) {
			debug('no lineup');
			this.epg.talk(['schedulesError', data.iden], {action: 'schedules', error: { message: 'headend required' }});
			return;
		}
		
		var remember = Object.assign({}, data);
		remember.total = 0;
		remember.limit = 4800;
		remember.skip = 0;
		remember.ratelimit = data.ratelimit || 500;
		
		var Channels = this.epg.Models.Channels;
		var Schedules = this.epg.Models.Schedules;
		
		// run the database items through a queue
		var queue = async.queue(store, remember.ratelimit);
		
		var done = function() {
			_this.epg.talk('schedulesDone', {remember});
		}
		queue.drain = () => {
			getChannels()
		};
		
		getChannels()
		
		function getChannels(skip, limit) {
			// defaults
			if(!skip) var skip = remember.skip;
			if(!limit) var limit = remember.limit;
			
			process.nextTick(() => {
				_this.epg.talk(['refreshGuide', 'schedules', remember.iden],'Get channels from database 4500 at a time... Total schedule records so far ' + remember.total);
			});
			
			Channels.find({where: { headend: data.lineup, active: true}, skip: skip, limit: limit }, (err, docs) => {
				if(err) {
					debug(err)
				}
				if(docs && docs.length > 0) {
					data.stations = docs.map((v) => { return { stationID: v.stationID } });
					remember.limit = limit;
					remember.skip = docs.length + remember.skip;
					_this.epg.talk(['refreshGuide', 'schedules', remember.iden],'Got ' + docs.length + ' active channels... Asking agent for schedules.');
					return grab();
				} else {
				
					debug('done grabbing schedules');
					done();
					
				}
			});
			//debug('EPG', this.epg)
			_this.epg.talk(['refreshGuide', 'schedules', remember.iden], 'Adding schedules in background');
		}
		// cache
		function store(result, finished) {
			
			if(typeof result === 'object') {
					var pro = result;
					var n = 0;
					pro.programs.forEach((p, k) => {
						
						p.headend = remember.lineup;
						p.stationID = pro.stationID;
						
						p.airDateTime = moment(p.airDateTime).unix();
						p.airDateEndTime = moment.unix(p.airDateTime).add(p.duration, 's').unix();
						Schedules.updateOrCreate({ stationID: p.stationID, airDateTime: p.airDateTime, headend: p.headend, programID: p.programID }, p, (err, newdoc) => {
							if(err) {
								debug('error adding schedule', err, remember, p.programID, newdoc);
								_this.epg.talk(data.iden,{ error: { message: 'Error adding program ' + p.programID + ' to database' }});
							}
							
							n++;
							remember.total++;
							
							if(n === pro.programs.length) {
								process.nextTick(() => {
									_this.epg.talk(['refreshGuide', 'programs', remember.iden],'Added ' + n + ' programs for station ' + pro.stationID)
								});
								return finished();
							
							} else {
								
								return true;
							
							}		
						});
					});
			} else {
				debug('bad result', pro);
				return finished();
			}			
		}
		
		function grab() {
			// change the iden so we catch the results instead of sending over the socket
			data.iden = hat();
			_this.epg.onTalk(data.iden, (data) => {
				queue.push(data, function (err) {
					//debug('finished processing item');
				});
			});
			_this.epg.agent.schedules(data);
		}			
	}
/** 
	 * schedules and programs should be called froma seperate process.  
	 * we communicate with this.epg.talk for the cli program and broadcast for the connected clients
	 * 
	 * */
schedules and programs should be called froma seperate process.
we communicate with this.epg.talk for the cli program and broadcast for the connected clients
schedules and programs should be called froma seperate process . we communicate with this . epg . talk for the cli program and broadcast for the connected clients
schedules( data ) {
		debug('listener schedules')
		var _this = this;
		if(isFalse(data.cache)) {
			this.epg.onTalk('schedules', (results) => {
			});
			return this.epg.agent.schedules(data);
		}
		if(!data.lineup) {
			debug('no lineup');
			this.epg.talk(['schedulesError', data.iden], {action: 'schedules', error: { message: 'headend required' }});
			return;
		}
		var remember = Object.assign({}, data);
		remember.total = 0;
		remember.limit = 4800;
		remember.skip = 0;
		remember.ratelimit = data.ratelimit || 500;
		var Channels = this.epg.Models.Channels;
		var Schedules = this.epg.Models.Schedules;
		var queue = async.queue(store, remember.ratelimit);
		var done = function() {
			_this.epg.talk('schedulesDone', {remember});
		}
		queue.drain = () => {
			getChannels()
		};
		getChannels()
		function getChannels(skip, limit) {
			if(!skip) var skip = remember.skip;
			if(!limit) var limit = remember.limit;
			process.nextTick(() => {
				_this.epg.talk(['refreshGuide', 'schedules', remember.iden],'Get channels from database 4500 at a time... Total schedule records so far ' + remember.total);
			});
			Channels.find({where: { headend: data.lineup, active: true}, skip: skip, limit: limit }, (err, docs) => {
				if(err) {
					debug(err)
				}
				if(docs && docs.length > 0) {
					data.stations = docs.map((v) => { return { stationID: v.stationID } });
					remember.limit = limit;
					remember.skip = docs.length + remember.skip;
					_this.epg.talk(['refreshGuide', 'schedules', remember.iden],'Got ' + docs.length + ' active channels... Asking agent for schedules.');
					return grab();
				} else {
					debug('done grabbing schedules');
					done();
				}
			});
			_this.epg.talk(['refreshGuide', 'schedules', remember.iden], 'Adding schedules in background');
		}
		function store(result, finished) {
			if(typeof result === 'object') {
					var pro = result;
					var n = 0;
					pro.programs.forEach((p, k) => {
						p.headend = remember.lineup;
						p.stationID = pro.stationID;
						p.airDateTime = moment(p.airDateTime).unix();
						p.airDateEndTime = moment.unix(p.airDateTime).add(p.duration, 's').unix();
						Schedules.updateOrCreate({ stationID: p.stationID, airDateTime: p.airDateTime, headend: p.headend, programID: p.programID }, p, (err, newdoc) => {
							if(err) {
								debug('error adding schedule', err, remember, p.programID, newdoc);
								_this.epg.talk(data.iden,{ error: { message: 'Error adding program ' + p.programID + ' to database' }});
							}
							n++;
							remember.total++;
							if(n === pro.programs.length) {
								process.nextTick(() => {
									_this.epg.talk(['refreshGuide', 'programs', remember.iden],'Added ' + n + ' programs for station ' + pro.stationID)
								});
								return finished();
							} else {
								return true;
							}		
						});
					});
			} else {
				debug('bad result', pro);
				return finished();
			}			
		}
		function grab() {
			data.iden = hat();
			_this.epg.onTalk(data.iden, (data) => {
				queue.push(data, function (err) {
				});
			});
			_this.epg.agent.schedules(data);
		}			
	}
schedules ( data ) { debug ( 'listener schedules' ) var _this = this ; if ( isFalse ( data . cache ) ) { this . epg . onTalk ( 'schedules' , ( results ) => { } ) ; return this . epg . agent . schedules ( data ) ; } if ( ! data . lineup ) { debug ( 'no lineup' ) ; this . epg . talk ( [ 'schedulesError' , data . iden ] , { action : 'schedules' , error : { message : 'headend required' } } ) ; return ; } var remember = Object . assign ( { } , data ) ; remember . total = 0 ; remember . limit = 4800 ; remember . skip = 0 ; remember . ratelimit = data . ratelimit || 500 ; var Channels = this . epg . Models . Channels ; var Schedules = this . epg . Models . Schedules ; var queue = async . queue ( store , remember . ratelimit ) ; var done = function ( ) { _this . epg . talk ( 'schedulesDone' , { remember } ) ; } queue . drain = ( ) => { getChannels ( ) } ; getChannels ( ) function getChannels ( skip , limit ) { if ( ! skip ) var skip = remember . skip ; if ( ! limit ) var limit = remember . limit ; process . nextTick ( ( ) => { _this . epg . talk ( [ 'refreshGuide' , 'schedules' , remember . iden ] , 'Get channels from database 4500 at a time... Total schedule records so far ' + remember . total ) ; } ) ; Channels . find ( { where : { headend : data . lineup , active : true } , skip : skip , limit : limit } , ( err , docs ) => { if ( err ) { debug ( err ) } if ( docs && docs . length > 0 ) { data . stations = docs . map ( ( v ) => { return { stationID : v . stationID } } ) ; remember . limit = limit ; remember . skip = docs . length + remember . skip ; _this . epg . talk ( [ 'refreshGuide' , 'schedules' , remember . iden ] , 'Got ' + docs . length + ' active channels... Asking agent for schedules.' ) ; return grab ( ) ; } else { debug ( 'done grabbing schedules' ) ; done ( ) ; } } ) ; _this . epg . talk ( [ 'refreshGuide' , 'schedules' , remember . iden ] , 'Adding schedules in background' ) ; } function store ( result , finished ) { if ( typeof result === 'object' ) { var pro = result ; var n = 0 ; pro . programs . forEach ( ( p , k ) => { p . headend = remember . lineup ; p . stationID = pro . stationID ; p . airDateTime = moment ( p . airDateTime ) . unix ( ) ; p . airDateEndTime = moment . unix ( p . airDateTime ) . add ( p . duration , 's' ) . unix ( ) ; Schedules . updateOrCreate ( { stationID : p . stationID , airDateTime : p . airDateTime , headend : p . headend , programID : p . programID } , p , ( err , newdoc ) => { if ( err ) { debug ( 'error adding schedule' , err , remember , p . programID , newdoc ) ; _this . epg . talk ( data . iden , { error : { message : 'Error adding program ' + p . programID + ' to database' } } ) ; } n ++ ; remember . total ++ ; if ( n === pro . programs . length ) { process . nextTick ( ( ) => { _this . epg . talk ( [ 'refreshGuide' , 'programs' , remember . iden ] , 'Added ' + n + ' programs for station ' + pro . stationID ) } ) ; return finished ( ) ; } else { return true ; } } ) ; } ) ; } else { debug ( 'bad result' , pro ) ; return finished ( ) ; } } function grab ( ) { data . iden = hat ( ) ; _this . epg . onTalk ( data . iden , ( data ) => { queue . push ( data , function ( err ) { } ) ; } ) ; _this . epg . agent . schedules ( data ) ; } }
schedules and programs should be called froma seperate process.
schedules and programs should be called froma seperate process .
// is cache false //this.emit(['schedules', data.iden], results); //if(data._broadcast) { //	this.broadcast(['schedules', data.iden], results); //}  // run the database items through a queue // defaults //debug('EPG', this.epg) // cache // change the iden so we catch the results instead of sending over the socket //debug('finished processing item');