/* $Id: nws_proxy.c,v 1.50 2004/12/02 04:10:54 graziano Exp $ */

#include "config_nws.h"

#include <stdio.h>
#include <string.h>

#include "diagnostic.h"
#include "protocol.h"
#include "osutil.h"
#include "dnsutil.h"
#include "messages.h"
#include "skills.h"
#include "strutil.h"
#include "nws_proxy.h"
#include "nws_api.h"

/* we have here the list of series (and respective memories) that we are
 * keeping track of. We don't use the timeout (in registrations) so we
 * use it to keep track of the index where we can find the right
 * forecaster. #getOld# contains the object-list of series we need to get
 * old datas for. #forecasts#
 * contains a list of forecasts (NULL for empty slot) and
 * #highestForecastIndex# is the length of the array. 
 * #lost# is the list of series we need to restart autofetching for.
 */
static registrations *series;
static ObjectSet toStart;
static Object upTo;
static char *getOld, *lost;
static int getOldLen, lostLen;
static NWSAPI_ForecastState **forecasts;
static int highestForecastIndex = 0;

/* performance counter */
static long updates = 0;

static void *lock = NULL;	 	/* lock for the module */


#if 0
/* remove the data and stop the autofetching for a single series */
static int
DeleteSingleForecast(	const char *seriesName) {
	int i;

	/* sanity check */
	if (seriesName == NULL) {
		WARN("DeleteSingleForecast: NULL parameter\n");
		return 0;
	}

	GetNWSLock(&lock);
	/* look for the registration */
	if (!SearchForName(series, seriesName, 1, &i)) {
		ReleaseNWSLock(&lock);
		LOG1("DeleteSingleForecast: no such series (%s)\n", seriesName);
		return 0;
	}
	/* delete forecaster and registration */
	NWSAPI_FreeForecastState(&forecasts[series->expirations[i]]);
	DeleteRegistration(series, i);
	ReleaseNWSLock(&lock);

	return 1;
}
#endif


/* add a forecaster */
static int
AddSingleForecast(	const Object seriesName) {
	int i;
	long j;
	char *name;

	/* sanity check */
	if (seriesName == NULL) {
		ERROR("AddSingleForecaster: NULL parameter.\n");
		return 0;
	}
	/* let's be sure it's a series and does have name attribute */
	name = NwsAttributeValue_r(FindNwsAttribute(seriesName, "objectClass"));
	if (name == NULL || strcmp(name, "nwsSeries")) {
		FREE(name);
		WARN("AddSingleForecaster: object is not a series!\n");
		return 0;
	}
	FREE(name);
	name = NwsAttributeValue_r(FindNwsAttribute(seriesName, "name"));
	if (name == NULL) {
		WARN("AddSingleForecaster: object with no name??\n");
		return 0;
	}
	FREE(name);

	/* since we are using index to global structure we should lock
	 * all the function */
	GetNWSLock(&lock);

	/* search for empty forecaster */
	for (j = 0; j < highestForecastIndex; j++) {
		if (forecasts[j] == NULL) {
			break;
		}
	}
	if (j >= highestForecastIndex) {
		/* make room for the forecaster */
		forecasts = REALLOC(forecasts, sizeof(NWSAPI_ForecastState **)*(highestForecastIndex + 1));
		if (forecasts == NULL) {
			ReleaseNWSLock(&lock);
			ERROR("AddSingleForecaster: out of memory\n");
			return 0;
		}
		j = highestForecastIndex;
		forecasts[j] = NULL;
		highestForecastIndex++;
	}

	/* let's see if we already have it */
	if (!SearchForObject(series, seriesName, &i)) {
		if (!InsertRegistration(series, seriesName, j, i)) {
			ReleaseNWSLock(&lock);
			ERROR1("AddSingleForecaster: failed to insert %s\n", seriesName);
			return 0;
		}
		forecasts[j] = NWSAPI_NewForecastState();
		if (forecasts[j] == NULL) {
			DeleteRegistration(series, i);
			ReleaseNWSLock(&lock);
			ERROR("AddSingleForecaster: out of memory\n");
			return 0;
		}
		series->expirations[i] = j;
	}
	ReleaseNWSLock(&lock);

	return 1;
}

#define HOW_MANY_FORECAST	(50)
static int
RestartForecast(	const char *name) {
	int i;
	unsigned int howMany;
	NWSAPI_Measurement meas[HOW_MANY_FORECAST];

	/* sanity check */
	if (name == NULL) {
		ERROR("RestartForecast: NULL parameter.\n");
		return 0;
	}

	/* since we are using index to global structure we should lock
	 * all the function */
	GetNWSLock(&lock);

	/* let's see if we already have it */
	if (!SearchForName(series, name, 1, &i)) {
		ReleaseNWSLock(&lock);
		ERROR1("RestartForecast: unknow series %s\n", name);
		return 0;
	}

	/* let's fetch the measurements */
	howMany = 0;
	if (!NWSAPI_GetMeasurements(name, NWSAPI_BEGINNING_OF_TIME, meas, HOW_MANY_FORECAST, &howMany)) {
		ReleaseNWSLock(&lock);
		ERROR1("RestartForecast: couldn't get measurements for %s\n", name);
		return 0;
	}

	/* let's restart the Forecaster */
	NWSAPI_FreeForecastState(&forecasts[series->expirations[i]]);
	forecasts[series->expirations[i]] = NWSAPI_NewForecastState();
	if (forecasts[series->expirations[i]] == NULL) {
		ReleaseNWSLock(&lock);
		ERROR("RestartForecast: out of memory\n");
		return 0;
	}

	/* let's update the forecaster state */
	NWSAPI_UpdateForecast(forecasts[series->expirations[i]], meas, howMany);
	ReleaseNWSLock(&lock);

	/* feed back for the new series */
	INFO1("RestartForecast: got old data for %s\n", name);

	return 1;
}

/* used by ProcessRequest to service GET_FORECASTS. We returns the
 * forecasts in #col# for #resource# (possibly specifying which options
 * we'd like to have) for the #howMany#-long list of #hosts#. We returned
 * in #filled# how many forecasters we filled. */
static int
GetAllForecasts(	char *resource,
			char *opts,
			char **hosts,
			int howMany,
			NWSAPI_ForecastCollection *col,
			int *filled) {
	int i, interMachine, j, z, len1, len2, ret, ind;
	char *name;
	
	/* sanity check */
	if (resource==NULL || hosts == NULL || col == NULL || filled == NULL) {
		ERROR("GetAllForecasts: NULL parameter(s)\n");
		return 0;
	}
	*filled = 0;
	if (NWSAPI_IsResourceValid(resource) == -1) {
		WARN1("GetAllForecasts: unknown resource (%s)\n", resource);
		return 0;
	}

	/* let's see if the resource is a 2 hosts resources (ie bw and
	 * lat) */
	interMachine = NWSAPI_IntermachineResource(resource);

	/* we don't parse options right now */
	if (opts != NULL || strlen(opts) > 0) {
		WARN("GetAllForecasts: we don't parse options right now!\n");
	}

	GetNWSLock(&lock);
	/* let's go through all the hosts */
	for (i = 0, ind = 0; i < howMany; i++) {
		for (j = 0; j < howMany; j++) {
			/* if we are asking for a single resource series,
			 * we don't need to go further then j = 0 */
			if (!interMachine && j > 0) {
				break;
			}

			/* set to not-good by default */
			col[ind].measurement.timeStamp = 0;
			(*filled)++;
			ind++;

			/* the series name will start with hosts[i] */
			if (!SearchForName(series, hosts[i], 0, &z)) {
				/* there is no such host */
				continue;
			}
			len1 = strlen(hosts[i]);
			len2 = strlen(hosts[j]);

			for (name = NULL; z >= 0; z--) {
				/* are we still talking about a possible
				 * series (initial name matches)? */
				name = NwsAttributeValue_r(FindNwsAttribute(series->vals[z], "name"));
				if (name == NULL) {
					ABORT("GetAllForecasts: out of memory\n");
				}

				/* let's check it still matches */
				ret = strncmp(name, hosts[i], len1);
				FREE(name);
				if (ret != 0) {
					/* wrong host */
					break;
				}

				/* let's see if the resource is right */
				name = NwsAttributeValue_r(FindNwsAttribute(series->vals[z], "resource"));
				if (name == NULL) {
					ABORT("GetAllForecasts: out of memory\n");
				}
				ret = strcmp(name, resource);
				FREE(name);
				if (ret != 0) {
					/* wrong resource */
					continue;
				}

				/* now let's see if we need to check the
				 * target */
				if (interMachine) {
					name = NwsAttributeValue_r(FindNwsAttribute(series->vals[z], "target"));
					if (name == NULL) {
						ABORT("GetAllForecasts: out of memory\n");
					}
					ret = strncmp(name, hosts[j], len2);
					FREE(name);
					if (ret != 0) {
						/* nope, wrong target */
						continue;
					}
				}

				/* got it! */
				if (!NWSAPI_ComputeForecast(forecasts[series->expirations[z]], &col[ind -1])) {
					ERROR("GetAllForecasts: failed to compute forecasts!\n");
				} else {
					/* done for this round */
					break;
				}
			}
		}
	}
	ReleaseNWSLock(&lock);

	return 1;
}

#define MY_FILTER "&(objectclass=nwsSeries)(resource="
static int
AskForAllSeries(	char *resource,
			char *opts,
			char **hosts) {
	int interMachine, i, j;
	char *filter;
	NWSAPI_ObjectSet whereTo;
	NWSAPI_Object obj;

	/* sanity check: opts can be NULL */
	if (resource == NULL || hosts == NULL || hosts[0] == NULL) {
		WARN("AskForAllSeries: NULL parameter\n");
		return 0;
	}

	interMachine = NWSAPI_IntermachineResource(resource);

	/* let's start to allocate some memory for the filter: this
	 * should be enough to get going. We'll realloc later on */
	filter = MALLOC(40 + strlen(resource));
	if (filter == NULL) {
		ERROR("AskForAllSeries: out of memory!!\n");
		return 0;
	}
	filter[0] = '\0';
	sprintf(filter, "&(objectclass=nwsSeries)(resource=%s)(|", resource);

	/* now we have to add all the hosts: if we have an
	 * interMachineResource we have to couple host=* and resource=*,
	 * otherwise host=* is sufficient */
	for (i = 0; hosts[i] != NULL; i++) {
		for (j = 0; hosts[j] != NULL; j++) {
			/* if we are on a single host resource we are
			 * done after after one iteration */
			if (!interMachine && j > 0) {
				break;
			}

			/* if we do interMachine, then skip the i = j */
			if (i == j && interMachine) {
				continue;
			}

			/* get same memory for the filter */
			filter = REALLOC(filter, strlen(filter) + strlen(hosts[i]) + strlen(hosts[j]) + 25);
			if (filter == NULL) {
				ERROR("AskForAllSeries: out of memory\n");
				return 0;
			}
			/* add the host= part */
			strcat(filter, "(&(host=");
			strcat(filter, hosts[i]);
			strcat(filter, "*");

			/* do we need the target= ?*/
			if (interMachine) {
				strcat(filter, ")(target=");
				strcat(filter, hosts[j]);
				strcat(filter, "*");
			}

			/* close this subclause */
			strcat(filter, "))");
		}
	}
	/* close the whole filter */
	strcat(filter, ")");

	/* let's query the nameserver */
	if (!NWSAPI_GetObjects(filter, &whereTo)) {
		ERROR("AskForAllSeries: failed to talk to the nameserver\n");
		FREE(filter);
		return 0;
	} else {
		/* let's add it to the todo list */
		NWSAPI_ForEachObject(whereTo, obj) {
			if (!NWSAPI_AddObject(&toStart, obj)) {
				ERROR("AskForAllSeries: out of memory!\n");
				break;
			}
		}
		NWSAPI_FreeObjectSet(&whereTo);
	}
	FREE(filter);

	return 1;
}

/* fills the resource, options and hosts coming from the net. Returns the
 * # of hosts, 0 on error */
static int
FillHosts(	Socket *sd,
		MessageHeader header,
		char **resource,
		char **opts,
		char ***h) {
	char *tmp, *ptr, **hosts;
	DataDescriptor des = SIMPLE_DATA(CHAR_TYPE, 0);
	int i;

	/* allocate memory */
	tmp = MALLOC(header.dataSize * sizeof(char));
	if (tmp == NULL) {
		return 0;
	}
	des.repetitions = header.dataSize;
	/* get all the data */
	if (!RecvData(*sd, tmp, &des, 1, -1)) {
		FREE(tmp);
		return 0;
	}
	tmp[header.dataSize] = '\0';
	ptr = tmp;

	/* first is the resource name */
	*resource = ptr;

	/* then the options */
	ptr += strlen(ptr) + 1;
	*opts = ptr;
	ptr += strlen(ptr) + 1;

	/* then all the hostnames: a lot of realloc down here */
	hosts = NULL;
	for(i = 0; (ptr - tmp + strlen(ptr) + 1) <= header.dataSize; i++) {
		/* allocate space for the pointer first */
		hosts = (char **)REALLOC(hosts, sizeof(char *)*(i + 2));
		if (hosts == NULL) {
			ABORT("FillHost: out of memory!");
		}
		hosts[i] = strdup(ptr);
		if (hosts[i] == NULL) {
			ABORT("FillHosts: out of memory!");
		}
		hosts[i + 1] = NULL;
		ptr += strlen(ptr) + 1;
	}
	/* check if we do indeed have hosts */
	if (hosts == NULL) {
		FREE(tmp);
		return 0;
	}
	*h = hosts;

	return i;
}
static void
ProcessRequest(	Socket *sd,
		MessageHeader header) {
	NWSAPI_Measurement meas;
	char name[255 + 1], *resource, *opts, **hosts;
	int i, j, howMany;
	NWSAPI_ForecastCollection *col;
	DataDescriptor howManyDes = SIMPLE_DATA(INT_TYPE, 1);

	switch(header.message) {
	case STATE_FETCHED:
		/* we just received and update */
		if (!NWSAPI_AutoFetchMessage(*sd, name, 256, &meas)) {
			ERROR("ProcessRequest: failed to autofetch!\n");
			break;
		}

		/* let's update */
		GetNWSLock(&lock);
		if (!SearchForName(series, name, 1, &i)) {
			WARN1("ProcessRequest: %s unknown series\n", name);
		} else {
			updates++;
			NWSAPI_UpdateForecast(forecasts[series->expirations[i]], &meas, 1);
			DDEBUG1("ProcessRequest: updates %s\n", name);
		}
		ReleaseNWSLock(&lock);
		
		break;

	case GET_FORECASTS:
		i = FillHosts(sd, header, &resource, &opts, &hosts);
		if (i == 0) {
			SendMessage(*sd, PROXY_FAILED, -1);
			ERROR("ProcessRequest: error in reading hosts!\n");
			DROP_SOCKET(sd);
			break;
		}

		/* let's allocate room for the forecasts */
		col = (NWSAPI_ForecastCollection *)MALLOC(sizeof(NWSAPI_ForecastCollection)*i*i);
		if (col == NULL) {
			SendMessage(*sd, PROXY_FAILED, -1);
			ERROR("ProcessRequest: out of memory!\n");
			DROP_SOCKET(sd);
			break;
		}

		/* now let's get the forecats */
		if (GetAllForecasts(resource, opts, hosts, i, col, &howMany)) {
			/* send how many data we are shipping */
			if (!SendMessageAndData(*sd, GOT_FORECASTS, &howMany, &howManyDes, 1, -1 )) {
				ERROR("ProcessRequest: failed to send GOT_FORECASTS\n");
				DROP_SOCKET(sd);
			} else for (j = 0; j < howMany; j++) {
				/* we need to send one forecast at a time */
				if (!SendData(*sd, &col[j], forecastCollectionDescriptor, forecastCollectionDescriptorLength, -1 )) {
					ERROR("ProcessRequest: failed to send GOT_FORECASTS\n");
					DROP_SOCKET(sd);
					break;
				}
			}
		} else {
			SendMessage(*sd, PROXY_FAILED, -1);
			DROP_SOCKET(sd);
		}
		for (;i >= 0; i--) {
			FREE(hosts[i]);
		}
		FREE(hosts);
		
		break;

	case START_FETCHING:
		i = FillHosts(sd, header, &resource, &opts, &hosts);
		if (i == 0) {
			SendMessage(*sd, PROXY_FAILED, -1);
			ERROR("ProcessRequest: error in reading hosts!\n");
			DROP_SOCKET(sd);
			break;
		}

		/* no error message to the client from here */
		SendMessage(*sd, PROXY_ACK, -1);

		/* get all the series names */
		AskForAllSeries(resource, opts, hosts);

		/* done: free memory */
		for (;i >= 0; i--) {
			FREE(hosts[i]);
		}
		FREE(hosts);

		break;
	
	default:
		ERROR1("ProcessRequest: unknown message %d\n", header.message);
		DROP_SOCKET(sd);
		break;
	}

	/* done with this socket */
	SocketIsAvailable(*sd);
}

static int
AddToTabList(	char **list,
		int len,
		const char *src) {
	int srcLen;

	srcLen = strlen(src);
	*list = REALLOC(*list, len + srcLen + 2);
	if (*list == NULL) {
		ERROR("out of memory\n");
		return 0;
	}
	if (len > 0) {
		(*list)[len] = '\t';
		len++;
	}
	memcpy(*list + len, src, srcLen + 1);
	len += srcLen;

	return len;
}

#define SWITCHES "AN:v:p:V"

void
usage() {
	printf("\nUsage: nws_proxy [OPTIONS]\n");
	printf("Proxy for the Network Weather Service\n");
	printf("\nOPTIONS can be:\n");
	printf("\t-N nameserver       use this nameserver\n");
	printf("\t-p port             bind to port instead of the default\n");
	printf("\t-A                  autofetch all the registered series\n");
	printf("\t-v level            verbose level (up to 5)\n");
	printf("\t-V                  print version\n");
	printf("Report bugs to <nws@nws.cs.ucsb.edu>.\n\n");
}


#define NO_MORE_THEN 1000

char * GenerateList(const registrations *r);

int
main (int argc, char **argv) {
	NWSAPI_HostSpec NS;
	int opt, verbose, len, startFetching;
	unsigned short port;
	extern char *optarg;
	extern int optind;
	FILE *logFD, *errFD;
	double now, wakeup, nextErrorCheck, nextBeatTime, nextGetOldData;
	char *tmp, name[255], *pidFile;
	ObjectSet mySet;

	/* default values  */
	pidFile = NULL;
	verbose = 2;
	logFD = stdout;
	errFD = stderr;
	len = 0;
	startFetching = 0;
	port = NWSAPI_GetDefaultPort(NWSAPI_PROXY_HOST);
	
	/* get the default nameserver */
	tmp = NWSAPI_EnvironmentValue_r("NAME_SERVER", "localhost");
	if (tmp == NULL) {
		ABORT("out of memory\n");
	}
	NS = *NWSAPI_MakeHostSpec(tmp, NWSAPI_GetDefaultPort(NWSAPI_NAME_SERVER_HOST));
	FREE(tmp);

	while((int)(opt = getopt(argc, argv, SWITCHES)) != EOF) {
		switch(opt) {
		case 'l':
			/* open the log file */
			logFD = fopen(optarg, "w");
			if (logFD == NULL) {
				printf("Couldn't open %s!\n", optarg);
				exit(1);
			}
			break;

		case 'e':
			/* open the error file */
			errFD = fopen(optarg, "w");
			if (errFD == NULL) {
				printf("Couldn't open %s!\n", optarg);
				exit(1);
			}
			break;

		case 'i':
			/* write pid to file */
			pidFile = strdup(optarg);
			if (pidFile == NULL) {
				ABORT("out of memory\n");
			}
			break;

		case 'A':
			startFetching = 1;
			break;

		case 'p':
			port = (unsigned short)strtol(optarg, NULL, 10);
			break;

		case 'N':
			NS = *NWSAPI_MakeHostSpec(optarg, NWSAPI_GetDefaultPort(NWSAPI_NAME_SERVER_HOST));
			break;

		case 'V':
			printf("nws_proxy for NWS version %s", VERSION);
#ifdef HAVE_PTHREAD_H
			printf(", with thread support");
#endif
#ifdef WITH_DEBUG
			printf(", with debug support");
#endif
			printf("\n\n");
			exit(0);
			break;

		case 'v':
			verbose = (unsigned short)atol(optarg);
			break;

		case '?':
			if (optopt == 'v') {
				/* using the first level */
				verbose = 1;
				break;
			}

		default:
			usage();
			exit(1);
			break;
		}
	}

	/* let's set the verbose evel */
	SetDiagnosticLevel(verbose, errFD, logFD);

	/* proud to serve */
	fclose(stdin);

	if (!EstablishAnEar(port, port, NULL, NULL)) {
		ABORT("failed to establish an ear!\n");
	}

	/* now that we've got the port, we can print the pid into the *
	 * pid file. We thus avoid to override pid files of running *
	 * nameservers */
	if (pidFile != NULL) {
		FILE *pidfile = fopen(pidFile, "w");
		if (!pidfile) {
			ABORT1("Can't write pidfile %s\n", pidFile);
		}
		fprintf(pidfile, "%d", (int)getpid());
		fclose(pidfile);
		free(pidFile);
	}

	/* initialize the series structure */
	GetNWSLock(&lock);
	if (!InitRegistrations(&series)) {
		ABORT("failed to init internal structure\n");
	}
	toStart = NewObjectSet();
	upTo = NULL;
	lost = getOld = NULL;
	lostLen = getOldLen = 0;

	/* let's set the nameServer */
	if (!NWSAPI_UseNameServer(&NS)) {
		WARN2("%s:%d is not a valid nameserver!\n", NS.machineName, NS.machinePort);
	} else if (startFetching) {
		/* let's get all the series names */
		if (!NWSAPI_GetObjectsTimeout("(objectclass=nwsSeries)", &toStart, 0)) {
			WARN("main: cannot get series!\n");
		}
	}
	ReleaseNWSLock(&lock);

	RegisterListener(STATE_FETCHED, "STATE_FETCHED,", &ProcessRequest);
	RegisterListener(GET_FORECASTS, "GET_FORECASTS,", &ProcessRequest);
	RegisterListener(START_FETCHING, "START_FETCHING,", &ProcessRequest);

	/* main service loop */
	wakeup = nextErrorCheck = nextBeatTime = nextGetOldData = 0;
	while (1) {
		now = CurrentTime();

		/* we might want to register with the nameserver here */
		if (nextBeatTime < now) {
			LOG1("main: we have %d series\n", series->howMany);

			/* some feedback */
			LOG1("got %d updates\n", updates);
			updates = 0;

			nextBeatTime = now + DEFAULT_HOST_BEAT/5;
		}

		if (nextErrorCheck < now) {
			/* let's try to get failed series */
			tmp = NWSAPI_AutoFetchError();
			if (tmp != NULL) {
				lostLen = AddToTabList(&lost, lostLen, tmp);
				if (lostLen == 0) {
					ERROR("failed to rescue lost series\n");
				}
				FREE(tmp);
			}
			nextErrorCheck = now + 20;
		}

		if (nextBeatTime > nextErrorCheck) {
			wakeup = nextErrorCheck;
		} else {
			wakeup = nextBeatTime;
		}
		wakeup -= now;


		/* we do add new series for up to 1 seconds to avoid
		 * becoming unresponsive to the memory autofetching
		 * messages */
		upTo = NextObject(toStart, upTo);
		mySet = NewObjectSet();
		while (upTo != NULL && (CurrentTime() - now) < 1) {
			/* let's create the forecaster */
			if (!AddSingleForecast(upTo)) {
				WARN("failed to add forecaster\n");
			} else {
				if (!AddObject(&mySet, upTo)) {
					ABORT("out of memory!\n");
				}

				/* add the series to the restart list */
				tmp = NwsAttributeValue_r(FindNwsAttribute(upTo, "name"));
				if (tmp == NULL) {
					ABORT("out of memory\n");
				}
				getOldLen = AddToTabList(&getOld, getOldLen, tmp);
				if (getOldLen == 0) {
					ABORT("out of memory!\n");
				}
				FREE(tmp);
			}
			upTo = NextObject(toStart, upTo);
			if (upTo == NULL) {
				/* well, we are done! */
				FreeObjectSet(&toStart);
				toStart = NewObjectSet();
				if (toStart == NULL) {
					ABORT("out of memory\n");
				}
				LOG("done adding new series\n");
			}
		}

		/* now let's deal with lost series: we need to add to the
		 * mySet the list of objects to restart. Notice that we
		 * are using the same now, hence we do either the new
		 * series or we rescue the lost one. */
		while (lostLen > 0 && (CurrentTime() - now) < 1) {
			GETWORD(name, lost, (const char **)&tmp);
			if (!SearchForName(series, name, 1, &opt)) {
				WARN1("non existent series (%s)??\n", name);
				continue;
			}
			NWSAPI_AddObject(&mySet, series->vals[opt]);

			/* remove the series we worked on */
			memmove(lost, tmp, lostLen - (tmp - lost) + 1);
			lostLen -= (tmp - lost);
			if (lostLen == 0) {
				FREE(lost);
				LOG("done recovering lost series\n");
			} else {
				/* let's come back quickly */
				wakeup = -1;
			}
		}
		/* let's start autofetching */
		if (strlen(mySet) > 5) {
			/* in case we already have these series, we stop
			 * and restart the autofetching */
			NWSAPI_AutoFetchEnd(mySet);
			if (!NWSAPI_AutoFetchBeginObject(mySet)) {
				LOG("failed to start AutoFetch\n");
			}

			/* since we started autofetch we better give some
			 * time to serve all the memory message */
			wakeup = 0;
		}
		FreeObjectSet(&mySet);

		/* now let's try to fill the forecasts with old data:
		 * this will overload the memory if we have a lot of
		 * series, so we do it no more then once a second. */
		if (getOldLen > 0 && nextGetOldData < now) {
			GETWORD(name, getOld, (const char **)&tmp);
			if (!RestartForecast(name)) {
				/* we got problem with restart: let's
				 * wait a bit longer before asking it
				 * again */
				nextGetOldData = CurrentTime() + 30;
			} else {
				nextGetOldData = CurrentTime() + 1;
			}
			memmove(getOld, tmp, getOldLen - (tmp - getOld) + 1);
			getOldLen -= (tmp - getOld);
			if (getOldLen == 0) {
				FREE(getOld);
				LOG("done getting old series values\n");
			}
			wakeup = 1;
		}

		/* let's serve the messages: we want to be responsive
		 * with the memory, so we try to serve all messages
		 * first, while we are ramping up our effort. */
		while(getOldLen > 0 && ListenForMessages(-1)) {
			;	/* nothing to do */
		}

		/* now we can do the normal stint */
		ListenForMessages((wakeup > 0) ? wakeup : -1);
	}

	return 1;
}