/* $Id: nws_api.c,v 1.19 2004/12/06 20:05:44 graziano Exp $ */

#include "config_nws.h"

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>

#include "diagnostic.h"
#include "host_protocol.h"
#include "messages.h"
#include "strutil.h"
#include "dnsutil.h"
#include "nws_sensor.h"
#include "nws_memory.h"
#include "register.h"
#include "osutil.h"
#include "skills.h"
#include "messages.h"
#include "nws_proxy.h"
#include "nws_api.h"

#include "nws_forecast_api.h"

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

/*
 * Information cached for auto-fetch series that we're tracking.  A connection
 * to the memory that holds the series and the (tab-deliminted) list of series
 * names that this memory is sending to us. #lostSeries# contains the
 * list of series that have been disconnected.
 */
typedef struct {
	struct host_cookie memory;
	registrations *series;
} AutoFetchInfo;
static registrations *lostSeries = NULL;


/*
 * Module globals.  #autoFetches# holds a list of memory connections that we're
 * using to track auto-fetched series, and #autoFetchCount# is the length of
 * this list.  The host_cookies contain connections to a default forecaster,
 * memory, and name server.  All these values are overridden by a call to one
 * of the Use*() functions.  We're supposed to pick the "best" forecaster and
 * memory if the client doesn't specify one, and there is no longer a default
 * name server, so we leave the initial values emtpy. 
 */
static AutoFetchInfo *autoFetches = NULL;
static unsigned int autoFetchCount = 0;

static struct host_cookie defaultMemory = {"", 0, NO_SOCKET};
static struct host_cookie *NSes = NULL;
static int howManyNS = 0;

/*
 * First of all the FORECAST stuff
 */
void
NWSAPI_FreeForecastState(NWSAPI_ForecastState **state) {
	FORECASTAPI_FreeForecastState((FORECASTAPI_ForecastState **)state);
}

const char *
NWSAPI_MethodName(unsigned int methodIndex) {
	return (FORECASTAPI_MethodName(methodIndex));
}

NWSAPI_ForecastState *
NWSAPI_NewForecastState() {
	return (NWSAPI_ForecastState *)FORECASTAPI_NewForecastState();
}

void
NWSAPI_UpdateForecastState(	NWSAPI_ForecastState *s,
				const NWSAPI_Measurement *m,
				unsigned int howMany,
				NWSAPI_ForecastCollection *f,
				unsigned int howManyF) {
	FORECASTAPI_UpdateForecastState((FORECASTAPI_ForecastState *)s, 
			(FORECASTAPI_Measurement *)m, 
			howMany, 
			(FORECASTAPI_ForecastCollection *)f, 
			howManyF);
}

int
NWSAPI_ComputeForecast(	NWSAPI_ForecastState *state,
			NWSAPI_ForecastCollection *forecast) {
	return FORECASTAPI_ComputeForecast((FORECASTAPI_ForecastState *)state, 
			(FORECASTAPI_ForecastCollection *)forecast);
}

/* END FORECAST */



/* utility function to get a tab-separated list of names from a
 * registrations structure. Returns a char * that needs to be freed */
static char *
GenerateList(const registrations *r) {
	char *tmp, *name;
	int len, i;

	/* let's generate the list */
	for (tmp = NULL, len = 0, i = 0;  i < r->howMany; i++) {
		name = NwsAttributeValue_r(FindNwsAttribute(r->vals[i], "name"));
		if (name == NULL) {
			WARN("GenerateList: no name??\n");
			continue;
		}
		tmp = REALLOC(tmp, len + strlen(name) + 2);
		if (tmp == NULL) {
			ABORT("GenerateList: out of memory\n");
		}
		if (len > 0) {
			tmp[len] = '\t';
			len++;
		}
		memcpy(tmp + len, name, strlen(name) + 1);
		len += strlen(name);
		FREE(name);
	}

	return tmp;
}

/* utility functions to add/remove a list of series from lostSeries
 * (which contains the series which aren't autofetching anymore due to
 * some errors) */
static void
AddToLost(const char *myList) {
	char name[255 + 1];
	const char *ind;
	int i;
	Object obj;

	/* quick check */
	if (myList == NULL || myList[0] == '\0') {
		return;
	}

	GetNWSLock(&lock);
	if (lostSeries == NULL) {
		if (!InitRegistrations(&lostSeries)) {
			ABORT("AddToLost: failed to initialize structure\n");
		}
	}
	for (ind = myList; GETWORD(name, ind, &ind); ) {
		if (SearchForName(lostSeries, name, 1, &i)) {
			/* we already have it */
			continue;
		}
		
		/* we need to insert it */
		obj = NewObject();
		AddNwsAttribute(&obj, "name", name);
		if (!InsertRegistration(lostSeries, obj, 0, i)) {
			WARN1("AddToLost: couldn't insert %s\n", name);
		}
		FreeObject(&obj);
	}
	LOG1("AddToLost: we have %d lost series\n", lostSeries->howMany);
	ReleaseNWSLock(&lock);

	return;
}
static void
AddToLostFromSocket(Socket sock) {
	int i;
	char * tmp;

	if (sock < 0) {
		return;
	}

	tmp = NULL;
	GetNWSLock(&lock);
	for (i = 0; i < autoFetchCount; i++) {
		if (autoFetches[i].memory.sd == sock) {
			/* found it */
			break;
		}
	}
	if (i < autoFetchCount) {
		tmp = GenerateList(autoFetches[i].series);
	}
	ReleaseNWSLock(&lock);

	AddToLost(tmp);
	FREE(tmp);

	return;
}
static void
RemoveFromLost(const char *myList) {
	char name[255 + 1];
	const char *ind;
	int i;

	/* sanity check */
	if (myList == NULL || lostSeries == NULL) {
		return;
	}

	GetNWSLock(&lock);
	for (ind = myList; GETWORD(name, ind, &ind); ) {
		if (!SearchForName(lostSeries, name, 1, &i)) {
			/* we don't have this series */
			continue;
		}
		DeleteRegistration(lostSeries, i);
	}
	LOG1("RemoveFromLost: we have %d lost series\n", lostSeries->howMany);
	ReleaseNWSLock(&lock);
}

/*
 * Disconnects the auto-fetch memory connected to #who# and removes its entry
 * from the #autoFetches# module global list.
 */
static void
AutoFetchDisconnect(Socket who) {
	int i;
	DataDescriptor des = SIMPLE_DATA(CHAR_TYPE, 1);

	/* just get out of here if no socket */
	if (who == NO_SOCKET) {
		return;
	}
	/* we operate on global variables all over the place */
	GetNWSLock(&lock);
	for (i = 0; i < autoFetchCount; i++) {
		if (autoFetches[i].memory.sd == who) {
			/* found it */
			break;
		}
	}
	if (i >= autoFetchCount) {
		ReleaseNWSLock(&lock);
		WARN("AutoFetchDisconnect: memory not found!\n");
		return;
	}

	INFO2("AutoFetchDisconnect: disconnecting from %s:%d\n", autoFetches[i].memory.name, autoFetches[i].memory.port);
	
	/* let's clean the autoFetches about this memory */
	while(autoFetches[i].series->howMany > 0) {
		DeleteRegistration(autoFetches[i].series, 0);
	}
	FREE(autoFetches[i].series);
	if (autoFetchCount > 1) {
		autoFetches[i] = autoFetches[--autoFetchCount];
	} else {
		FREE(autoFetches);
		autoFetchCount = 0;
	}
	ReleaseNWSLock(&lock);

	/* let's stop autofetch (empty string) */
	des.repetitions = 1;
	if (!SendMessageAndData(who, AUTOFETCH_BEGIN, "", &des, 1, 5)) {
		ERROR("AutoFetchDisconnect: failed to talk to memory\n");
	}

	/* Ignore the reply.  It might get mixed with an autofetch anyway */
	DROP_SOCKET(&who);
}

/*
 * Connects to #memory#, sends it #seriesList# in an AUTOFETCH_BEGIN message,
 * and places an entry for the connection in the #autoFetches# global.  Returns
 * 1 if successful, else 0.
 *
 * NOTE: we need to hook a check whenever a socket gets disconnected,
 * since it can be an autofetching socket.
 */
static int
AutoFetchConnect(	struct host_cookie *memory,
			const char *seriesList) {
	AutoFetchInfo *extendedAutoFetches;
	size_t ignored;
	DataDescriptor seriesDescriptor = SIMPLE_DATA(CHAR_TYPE, 0);
	char name[256], *series;
	const char *ind;
	Object obj;
	int i;

	/* sanity check */
	if (memory == NULL || seriesList == NULL) {
		FAIL("AutoFetchConnect: NULL parameters!\n");
	}

	/* let's check if we can talk to the memory first */
	if (!ConnectToHost(memory, NULL)) {
		FAIL1("AutoFetchConnect: Unable to contact memory %s\n", HostCImage(memory));
	}

	/* since we operates on the global variables, we lock a lot */
	GetNWSLock(&lock);
	/* extend the autoFetchers */
	extendedAutoFetches = (AutoFetchInfo *) REALLOC(autoFetches, (autoFetchCount + 1) * sizeof(AutoFetchInfo));
	if (extendedAutoFetches == NULL) {
		ReleaseNWSLock(&lock);
		FAIL("AutoFetchConnect: out of memory\n");
	}
	extendedAutoFetches[autoFetchCount].series = NULL;
	if (!InitRegistrations(&extendedAutoFetches[autoFetchCount].series)) {
		FREE(extendedAutoFetches);
		ReleaseNWSLock(&lock);
		FAIL("AutoFetchConnect: cannot initialize structure\n");
	}
	autoFetches = extendedAutoFetches;

	/* let's save the series/memory */
	autoFetches[autoFetchCount].memory = *memory;
	for (ind = seriesList; GETWORD(name, ind, &ind); ) {
		if (SearchForName(autoFetches[autoFetchCount].series, name, 1, &i)) {
			/* we already have it */
			continue;
		}

		/* insert the element */
		obj = NewObject();
		AddNwsAttribute(&obj, "name", name);
		if (!InsertRegistration(autoFetches[autoFetchCount].series, obj, 0, i)) {
			WARN1("AutoFetchConnect: couldn't insert %s\n", name);
		}
		FreeObject(&obj);
	}
	/* generate the list of series we got in */
	series = GenerateList(autoFetches[autoFetchCount].series);
	autoFetchCount++;
	ReleaseNWSLock(&lock);
	if (series == NULL) {
		WARN("AutoFetchConnect: empty series list?\n");
		return 0;
	}

	/* let's try to autofetch the series list */
	i = 1;
	seriesDescriptor.repetitions = strlen(series) + 1;
	if (!SendMessageAndData(memory->sd, AUTOFETCH_BEGIN, (const void *)series, &seriesDescriptor, 1, -1) || !RecvMessage(memory->sd, AUTOFETCH_ACK, &ignored, -1)) {
		i = 0;
		AutoFetchDisconnect(memory->sd);
		ERROR1("AutoFetchConnect: Memory %s refused request\n", HostCImage(memory));
	}
	FREE(series);

	return i;
}


/*
 * Contacts the name server to look up the series #seriesName# and return the
 * memory which is storing its measurements.  Returns 1 if successful, else 0.
 */
static int
GetSeriesMemory(	const char *seriesName,
                	struct host_cookie *memory,
			int timeout) {
	ObjectSet seriesObject;
	Object obj;
	char myFilter[255+1], *tmp; 
	int ret;

	snprintf(myFilter, sizeof(myFilter), "(name=%s)", seriesName);

	seriesObject = NewObjectSet();
	if (!NWSAPI_GetObjectsTimeout(myFilter, &seriesObject, timeout)) {
		FreeObject(&seriesObject);
		FAIL1("GetSeriesMemory: couldn't find the memory for %s\n", seriesName);
	}

	/* now we can have multiple series: we pick the one that does
	 * have a 'activity' field, since it means that is been
	 * currently updated. */
	ret = 0;
	NWSAPI_ForEachObject(seriesObject, obj) {
		tmp = NwsAttributeValue_r(FindNwsAttribute(obj, "memory"));
		if (tmp == NULL) {
			ERROR1("GetSeriesMemory: %s has no memory!\n", obj);
			continue;
		}

		Host2Cookie(tmp, DefaultHostPort(MEMORY_HOST), memory);
		FREE(tmp);
		ret = 1;	/* we indeed found something */

		/* if this series has an activity, we are done */
		if (FindNwsAttribute(obj, "activity") == NO_OBJECT) {
			break;
		}
	}
	FreeObject(&seriesObject);

	return ret;
}


#define OPTION_NAME_LEN (31 + 1)
typedef struct {
	char name[OPTION_NAME_LEN];
	unsigned int minOccurrences;
	unsigned int maxOccurrences;
} OptionInfo;


/*
 * A "local" function of StartActivity().  If #attr# is listed in #options#,
 * sets the fields of #intoWhere# based on the name and value of #attr# and
 * returns 1; otherwise, returns 0.
 */
static int
ParseOption(	const NWSAPI_NwsAttribute attr,
		const char *options,
		OptionInfo *intoWhere) {
	char *attrName, *tmp;
	OptionInfo newOption;

	attrName = NwsAttributeName_r(attr);
	if(attrName == NULL || strstr(options, attrName) == 0) {
		FREE(attrName);
		return 0;
	}

	/* Option values have the format MIN_to_MAX_TYPE */
	SAFESTRCPY(newOption.name, attrName);
	FREE(attrName);

	tmp = NwsAttributeValue_r(attr);
	if (tmp == NULL) {
		return 0;
	}

	/* find the min/max occurrences for this option */
	newOption.minOccurrences = (unsigned int)strtol(tmp, &attrName, 10);
	attrName += strlen("_to_");
	newOption.maxOccurrences = (unsigned int)strtol(attrName, NULL, 10);

	/* done */
	*intoWhere = newOption;
	FREE(tmp);

	return 1;
}


/*
 * given a #seriesSpec# returns the default options in a string (to be
 * freed).
 */
static char *
ResourceOptions(	const NWSAPI_SeriesSpec *series) {
	struct host_cookie host;
	const unsigned short tmp = DefaultHostPort(SENSOR_HOST);
	int res, skill, len;
	const MeasuredResources *r;
	const char *opts;
	char opt[35], options[EXP_LIST_SIZE]; 

	/* let's find first which resource we have */
	for (res = 0; res < RESOURCE_COUNT; res++) {
		if (strcmp(series->resourceName, ResourceName((MeasuredResources) res)) == 0) {
			break;
		}
	}
	if (res >= RESOURCE_COUNT) {
		/* nothing more we can do here: no idea which resource we
		 * are talking about */
		WARN1("ResourceOptions: unknown resource (%s)\n", series->resourceName);
		return NULL;
	}

	/* now let's find which skill is responsible for this resource */
	for (skill = 0; skill < SKILL_COUNT; skill++) {
		if (!SkillResources((KnownSkills)skill, &r, &len)) {
			/* ?? we shouldn't be here */
			continue;
		}
		for (; len > 0; len--) {
			if (r[len - 1] == (MeasuredResources) res) {
				/* found it */
				break;
			}
		}
		if (len > 0) {
			/* we found it */
			break;
		}
	}
	if (skill >= SKILL_COUNT) {
		/* nothing more we can do here: no idea which skill we
		 * are talking about */
		WARN1("ResourceOptions: unknown skill for resource (%s)\n", series->resourceName);
		return NULL;
	}

	/* now we can find the options supported by the skill, and find
	 * their deafaults */
	opts = SkillSupportedOptions((KnownSkills) skill);
	options[0] = '\0';
	while(GETTOK(opt, opts, ",", &opts)) {
		char *str;

		str = SkillOptionDefault((KnownSkills)skill, opt);
		if (opt == NULL) {
			/* nothing here */
			continue;
		}

		/* add to the options */
		if (options[0] != '\0') {
			strcat(options, "\t");
		}
		strcat(options, opt);
		strcat(options, ":");
		strcat(options, str);
		free(str);
	}

	/* do we have a destination? */
	if (series->destinationMachine[0] != '\0') {
		if (options[0] != '\0') {
			strcat(options, "\t");
		}
		strcat(options, "target:");
		Host2Cookie(series->destinationMachine, tmp, &host);
		strcat(options, HostCImage(&host));
	}

	return strdup(options);
}


/*
 * Attempts to connect to #host#.  If successful, copies host name and
 * port number into #whichCookie# and returns 1; else returns 0.
 */
static int
UseHost(	const NWSAPI_HostSpec *host,
		struct host_cookie *whichCookie) {
	struct host_cookie tmp;

	MakeHostCookie(host->machineName, host->machinePort, &tmp);
	if (!ConnectToHost(&tmp, NULL)) {
		FAIL1("Unable to contact %s\n", HostCImage(&tmp));
	}
	SocketInUse(tmp.sd);
	*whichCookie = tmp;

	return 1;
}


char *
NWSAPI_NwsAttributeName_r(NWSAPI_NwsAttribute attribute) {
	return NwsAttributeName_r(attribute);
}


char *
NWSAPI_NwsAttributeValue_r(NWSAPI_NwsAttribute attribute) {
	return NwsAttributeValue_r(attribute);
}


/* DEPRECATED */
const char *
NWSAPI_NwsAttributeName(NWSAPI_NwsAttribute attribute) {
	ERROR("NwsAttributeName has been deprecated: use the _r version\n");
	return NwsAttributeName(attribute);
}
const char *
NWSAPI_NwsAttributeValue(NWSAPI_NwsAttribute attribute) {
	ERROR("NwsAttributeValue has been deprecated: use the _r version\n");
	return NwsAttributeValue(attribute);
}
const char *
NWSAPI_EnvironmentValue(const char *name,
                        const char *defaultValue) {
	static char *ret = NULL;

	ERROR("EnvironmentValue has been deprecated: use the _r version\n");
	if (ret != NULL) {
		FREE(ret);
	}
	ret = GetEnvironmentValue(name, "~", ".nwsrc", defaultValue);

	return ret;
}
/* end deprecated */

char *
NWSAPI_AutoFetchError() {
	char *ret;
	int i;
	Socket s;

	/* first of all let's check that all the sockets we have here are
	 * good to go */
	GetNWSLock(&lock);
	for (i = 0; i < autoFetchCount; i++) {
		if (!IsOkay(autoFetches[i].memory.sd)) {
			s = autoFetches[i].memory.sd;
			WARN1("AutoFetchError: lost contact with memory %s\n", autoFetches[i].memory.name);
			ReleaseNWSLock(&lock);
			AddToLostFromSocket(s);
			AutoFetchDisconnect(s);
			GetNWSLock(&lock);

			/* let's recheck this index */
			i--;
		}
	}

	if (lostSeries == NULL) {
		ret = NULL;
	} else {
		ret = GenerateList(lostSeries);
		while (lostSeries->howMany > 0) {
			DeleteRegistration(lostSeries, 0);
		}
	}
	ReleaseNWSLock(&lock);

	return ret;
}

int
NWSAPI_AutoFetchAlready(const char *seriesName) {
	int i, ret;
	
	/* sanity check */
	if (seriesName == NULL) {
		FAIL("AutoFetchAlready: NULL parameter!\n");
	}

	/* easy way out */
	if (seriesName[0] == '\0') {
		return 0;
	}

	GetNWSLock(&lock);
	/* let's see if we already have this series */
	for(i = 0; i < autoFetchCount; i++) {
		if (SearchForName(autoFetches[i].series, seriesName, 1, &ret)) {
			break;
		}
	}
	if (i < autoFetchCount) {
		ret = 1;
	} else {
		ret = 0;
	}
	ReleaseNWSLock(&lock);

	return ret;
}

/* temporary structure used to keep track of which memory has which
 * series. The 2 following functions (AddMemorySeries & StartFetch) are
 * local to the AutoFetchBegin* functions.*/
typedef struct {
	struct host_cookie memory;
	char *toFetch;
} memorySeries;
static int
AddMemorySeries(	memorySeries **l,
			int *howMany,
			struct host_cookie *memory) {
	int i;
	memorySeries *list;

	list = *l;

	/* See whether we're already have this memory for other series. */
	for(i = 0; i < *howMany; i++) {
		if (SameHost(memory, &list[i].memory)) {
			break;
		}
	}
	if (i >= *howMany) {
		/* nope it's a new memory */
		i = *howMany;
		(*howMany)++;

		/* allocate space to save the memory */
		list = REALLOC(list, sizeof(memorySeries)*(*howMany));
		if (list == NULL) {
			return -1;
		}
		list[i].toFetch = strdup("");
		if (list[i].toFetch == NULL) {
			return -1;
		}

		list[i].memory = *memory;
		*l = list;
	}

	return i;
}
static int
StartFetch(	memorySeries *list,
		int howMany) {
	int i, j, len, ret;
	char *listToFetch, *c;
	Socket sock;

	ret = 0;

	if (howMany <= 0) {
		/* I guess we are done! */
		return 1;
	}
	for (i = 0; i < howMany; i++) {
		sock = NO_SOCKET;

		GetNWSLock(&lock);
		/* See whether we're already asking this memory for other
		 * series. */
		for(j = 0; j < autoFetchCount; j++) {
			if (SameHost(&list[i].memory, &autoFetches[j].memory)) {
				break;
			}
		}
		if (j >= autoFetchCount) {
			/* this is the first series for this memory */
			listToFetch = strdup(list[i].toFetch);
		} else {
			/* An existing memory.  Because we're receiving
			 * messages asynchronously, we can't reliably
			 * reuse the connection -- we might mistake the
			 * ack for a series coming in or vice versa.
			 * Instead, get rid of this connection and open a
			 * new one with the extended list.  */
			c = GenerateList(autoFetches[i].series);
			len = strlen(c) + strlen(list[i].toFetch) + 2;
			listToFetch = (char *)MALLOC(len);
			if (listToFetch) {
				len = strlen(c);
				memcpy(listToFetch, c, len);
				memcpy(listToFetch + len, list[i].toFetch, strlen(list[i].toFetch));
				sock = autoFetches[j].memory.sd;
			}
			FREE(c);
		}

		/* let's restart the autofecthing for this memory */
		ReleaseNWSLock(&lock);
		if (listToFetch == NULL) {
			ERROR("StartFetch out of memory\n");
			break;
		}
		if (sock != NO_SOCKET) {
			AutoFetchDisconnect(sock);
		}

		if (AutoFetchConnect(&list[i].memory, listToFetch)) {
			ret = 1;
		} else {
			AddToLost(listToFetch);
		}
		FREE(listToFetch);
	}

	/* let's free memory */
	for (i = 0; i < howMany; i++) {
		FREE(list[i].toFetch);
	}
	FREE(list);

	return ret;
}

int
NWSAPI_AutoFetchBegin(const char *seriesName) {
	int i, len;
	struct host_cookie memory;
	char tmpSeries[255];
	const char *ind;
	memorySeries *list;
	int howMany;

	/* sanity check */
	if (seriesName == NULL) {
		FAIL("AutoFetchBegin: NULL parameter\n");
	}

	/* let's loop through the series */
	list = NULL;
	for (howMany = 0, ind = seriesName; GETWORD(tmpSeries, ind, &ind); ) {
		/* let's see if we already checked this series */
		if (NWSAPI_AutoFetchAlready(tmpSeries)) {
			continue;
		}

		/* let's see if we can find the memory responsible */
		if (!GetSeriesMemory(tmpSeries, &memory, -1)) {
			WARN("AutoFetchBegin: unable to determine memory\n");
			AddToLost(tmpSeries);
			continue;
		}

		/* let's get the right index for this memory */
		i = AddMemorySeries(&list, &howMany, &memory);
		if (i == -1 ) {
			ABORT("AutoFetchBegin: out of memory\n");
			return 0;
		}

		/* now let's add the series name to the to be fetch list */
		len = strlen(tmpSeries) + strlen(list[i].toFetch) + 2;
		list[i].toFetch = REALLOC(list[i].toFetch, len);
		if (list[i].toFetch == NULL) {
			ERROR("AutoFetchBegin: out of memory\n");
			AddToLost(seriesName);
			for (i = 0; i < howMany; i++) {
				FREE(list[i].toFetch);
			}
			FREE(list);

			return 0;
		}
		strcat(list[i].toFetch, "\t");
		strcat(list[i].toFetch, tmpSeries);
	}

	/* we got all the memories/series coupled: time to fetch */
	return StartFetch(list, howMany);
}


int
NWSAPI_AutoFetchBeginObject(ObjectSet seriesSet) {
	int i, len;
	char *name;
	struct host_cookie memory;
	memorySeries *list;
	int howMany;
	Object obj;

	/* sanity check */
	if (seriesSet == NULL) {
		FAIL("AutoFetchBeginObject: NULL parameter\n");
	}

	/* let's loop through the series */
	list = NULL;
	obj = NWSAPI_FirstObject(seriesSet);
	for (howMany = 0; obj != NULL; obj = NextObject(seriesSet, obj)) {
		/* now we can get the name and the memory */
		name = NwsAttributeValue_r(FindNwsAttribute(obj, "memory"));
		if (name == NULL) {
			ERROR("AutoFetchBeginObject: series with no memory!\n");
			continue;
		}
		Host2Cookie(name, DefaultHostPort(NAME_SERVER_HOST), &memory);
		FREE(name);
		name = NwsAttributeValue_r(FindNwsAttribute(obj, "name"));
		if (name == NULL) {
			ERROR("AutoFetchBeginObject: series with no name!\n");
			continue;
		}

		/* let's see if we already checked this series */
		if (NWSAPI_AutoFetchAlready(name)) {
			FREE(name);
			continue;
		}

		/* let's get the right index for this memory */
		i = AddMemorySeries(&list, &howMany, &memory);
		if (i == -1 ) {
			ABORT("AutoFetchBegin: out of memory\n");
			return 0;
		}

		/* now let's add the series name to the to be fetch list */
		len = strlen(name) + strlen(list[i].toFetch) + 2;
		list[i].toFetch = REALLOC(list[i].toFetch, len);
		if (list[i].toFetch == NULL) {
			ERROR("AutoFetchBeginObject: out of memory\n");
			AddToLost(name);
			FREE(name);
			for (i = 0; i < howMany; i++) {
				FREE(list[i].toFetch);
			}
			FREE(list);

			return 0;
		}
		strcat(list[i].toFetch, "\t");
		strcat(list[i].toFetch, name);
		FREE(name);
	}

	/* we got all the memories/series coupled: time to fetch */
	return StartFetch(list, howMany);
}

int
NWSAPI_AutoFetchMessage(	int sock,
				char *seriesName,
				unsigned int nameLen,
				NWSAPI_Measurement *meas) {
	char *autoFetchContents;
	struct nws_memory_state s;
	const char *nextWord;
	DataDescriptor des = SIMPLE_DATA(CHAR_TYPE, 0);
	char measurement[127 + 1];
	char timeStamp[127 + 1];
	Socket sender;

	nextWord = autoFetchContents = NULL;

	/* sanity check */
	sender = (Socket)sock;
	if (sender == NO_SOCKET) {
		FAIL("AutoFetchMessage: no valid socket\n");
	}

	/* let's get the state */
	if (!RecvData(sender, &s, stateDescriptor, stateDescriptorLength, -1)) {
		AddToLostFromSocket(sender);
		AutoFetchDisconnect(sender);
		FAIL("AutoFetchMessage: failed to receive state\n");
	}

	/* get room for the data */
	des.repetitions = s.rec_size * s.rec_count;
	autoFetchContents = (char *)MALLOC(des.repetitions + 1);
	if (autoFetchContents == NULL) {
		AddToLostFromSocket(sender);
		AutoFetchDisconnect(sender);
		FAIL("AutoFetchMessage: out of memory\n");
	}

	/* let's get the message */
	if (!RecvData(sender, autoFetchContents, &des, 1, -1)) {
		AddToLostFromSocket(sender);
		AutoFetchDisconnect(sender);
		FREE(autoFetchContents);
		FAIL("AutoFetchMessage: failed to receive message\n");
	}

	/* let's parse it */
	if (GETWORD(timeStamp, autoFetchContents, &nextWord) && GETWORD(measurement, nextWord, &nextWord)) {
		/* let's return the values and series name */
		zstrncpy(seriesName, s.id, nameLen);
		meas->timeStamp = strtod(timeStamp, NULL);
		meas->measurement = strtod(measurement, NULL);
	} else {
		AddToLostFromSocket(sender);
		AutoFetchDisconnect(sender);
		FREE(autoFetchContents);
		FAIL("AutoFetchMessage: failed to parse message\n");
	}
	FREE(autoFetchContents);

	return 1;
}


int
NWSAPI_AutoFetchCheck(char *seriesName,
                      unsigned int nameLen,
                      NWSAPI_Measurement *seriesMeasurement,
                      long timeOut) {
	MessageHeader header;
	Socket sender;
	int i;

	/* sanity check */
	if (seriesName == NULL || seriesMeasurement == NULL || nameLen <= 0) {
		FAIL("AutoFetchCheck: wrong parameter!\n");
	}
	seriesName[0] = '\0';

	 /* let's wait for a message: a message contains only 1 series
	  * name plus 1 measurements */
	if (!IncomingRequest(timeOut, &sender)) {
		/* nope, nothing for us */
		return 0;
	}

	/* let's get the header */
	if (!RecvHeader(sender, &header, -1) ||
			header.message != STATE_FETCHED) {
		char *name;
		
		AddToLostFromSocket(sender);
		AutoFetchDisconnect(sender);

		name = PeerName_r(sender);
		ERROR1("AutoFetchCheck: failed to receive from %s\n", name);
		FREE(name);

		return 0;
	}

	/* let's get the auto fetch data */
	i = NWSAPI_AutoFetchMessage(sender, seriesName, nameLen, seriesMeasurement);

	/* done with this socket */
	SocketIsAvailable(sender);

	return i;
}


void
NWSAPI_AutoFetchEnd(const char *seriesName) {
	int i, len, j;
	struct host_cookie memory;
	char *tmp, name[255];
	const char *ind;
	Socket sock;

	/* sanity check */
	if (seriesName == NULL) {
		ERROR("NWSAPI_AutoFetchEnd: NULL parameter\n");
		return;
	}

	/* we can have multiple series: let's get them one by one */
	for (ind = seriesName; GETWORD(name, ind, &ind); ) {
		/* if we don't have it, we don't need to stop it */
		if (!NWSAPI_AutoFetchAlready(name)) {
			continue;
		}

		len = strlen(name);

		/* let's look for the series */
		GetNWSLock(&lock);
		for(sock = NO_SOCKET, tmp = NULL, i = 0; i < autoFetchCount; i++) {
			/* let's look for the series */
			if (!SearchForName(autoFetches[i].series, name, 1, &j)) {
				/* not here */
				continue;
			}

			/* let's remmeber the memory to restart */
			SAFESTRCPY(memory.name, autoFetches[i].memory.name);
			memory.port = autoFetches[i].memory.port;
			sock = autoFetches[i].memory.sd;

			/* let's remove it: if there is only one series
			 * we remove the whole memory. */
			if (autoFetches[i].series->howMany > 1) {
				DeleteRegistration(autoFetches[i].series, j);
				tmp = GenerateList(autoFetches[i].series);
			}

			/* since we found the series is time to reset the
			 * autofetching with the memory */
			break;
		}
		ReleaseNWSLock(&lock);

		if (sock != NO_SOCKET) {
			/* See comment in AutoFetchBegin() as to why we
			 * can't reuse the connection. */
			AutoFetchDisconnect(sock);
			memory.sd = NO_SOCKET;
			if (tmp != NULL && strlen(tmp) > 0) {
				if (!AutoFetchConnect(&memory, tmp)) {
					/* got problem: add the series to the
					 * orphaned series */
					AddToLost(tmp);
				}
			}
		}
		FREE(tmp);
	}

	/* we remove the series from the lostSeries no matter what: we
	 * don't care about it anymore */
	RemoveFromLost(seriesName);
}


char *
NWSAPI_EnvironmentValue_r(const char *name,
                        const char *defaultValue) {
	return GetEnvironmentValue(name, "~", ".nwsrc", defaultValue);
}

unsigned short
NWSAPI_GetDefaultPort(unsigned int hostType) {
	return DefaultHostPort(hostType);
}


int
NWSAPI_HaltActivity(	const NWSAPI_HostSpec *whichSensor,
			const char *activityName) {
	DataDescriptor activityDescriptor = SIMPLE_DATA(CHAR_TYPE, 0);
	size_t ignored;
	int returnValue;
	struct host_cookie tmp;

	/* sanity check */
	if (whichSensor == NULL || activityName == NULL) {
		FAIL("HaltActivity: NULL parameter(s)\n");
	}

	Host2Cookie(whichSensor->machineName, whichSensor->machinePort, &tmp);
	if (!ConnectToHost(&tmp, NULL)) {
		FAIL1("HaltActivity: unable to contact %s\n", HostCImage(&tmp));
	}

	activityDescriptor.repetitions = strlen(activityName) + 1;
	returnValue = SendMessageAndData(tmp.sd,
                                   ACTIVITY_STOP,
                                   activityName,
                                   &activityDescriptor,
                                   1,
                                   -1) &&
		RecvMessage(tmp.sd,
                            ACTIVITY_STOPPED,
                            &ignored,
                            -1);
	DisconnectHost(&tmp);

	return returnValue;
}


int
NWSAPI_AddObject(	NWSAPI_ObjectSet *toSet,
			const NWSAPI_Object obj) {
	return AddObject(toSet, obj);
}

NWSAPI_ObjectSet
NWSAPI_NewObjectSet() {
	return NewObjectSet();
}


void
NWSAPI_FreeObjectSet(NWSAPI_ObjectSet *toBeFreed) {
	FreeObjectSet(toBeFreed);
}



int
NWSAPI_GetHostsTimeout(	const char *filter,
			NWSAPI_HostSpec *whereTo,
			unsigned int atMost,
			unsigned int *numberReturned,
			int timeout) {
	char *c, *name;
	char *hostFilter;
	int i, port;
	struct host_cookie host;
	NWSAPI_Object    hostObject;
	NWSAPI_ObjectSet hostObjects;

	/* sanity check */
	if (filter == NULL || whereTo == NULL || numberReturned == NULL) {
		FAIL("GetHosts: NULL parameter(s)!\n");
	}

	hostFilter = (char *)MALLOC(strlen(filter) + strlen(NWSAPI_HOSTS) + 6);
	if (hostFilter == NULL) {
		FAIL("GetHosts: out of memory\n");
	}
	sprintf(hostFilter, "(&(%s)%s)", filter, NWSAPI_HOSTS);
	if (!NWSAPI_GetObjectsTimeout(hostFilter, &hostObjects, timeout)) {
		FREE(hostFilter);
		FAIL("GetHosts: object retrieval failed\n");
	}
	FREE(hostFilter);

	i = 0;
	NWSAPI_ForEachObject(hostObjects, hostObject) {
		if (i >= atMost) {
			break;
		}

		port = 0;
		name = NwsAttributeValue_r(FindNwsAttribute(hostObject, NWSAPI_PORT_ATTR));
		if (name == NULL) {
			WARN("NWSAPI_GetHosts: inconsistent host object\n");
			continue;
		}
		port = (int)strtol(name, NULL, 10);
		FREE(name);
		name = NwsAttributeValue_r(FindNwsAttribute(hostObject, NWSAPI_IP_ATTR));
		if (name == NULL) {
			WARN("NWSAPI_GetHosts: inconsistent host object\n");
			continue;
		}
		/* TBD: We really should iterate through the IP addresses
		 * here looking for one we can connect to.  Instead, we
		 * just grab the first one.  */
		if((c = strchr(name, ',')) != NULL) {
			*c = '\0';
		}

		/* let's normalize the names */
		Host2Cookie(name, port, &host);
		FREE(name);
		SAFESTRCPY(whereTo[i].machineName, host.name);
		whereTo[i].machinePort = host.port;
		i++;
	}
	NWSAPI_FreeObjectSet(&hostObjects);
  
	if (numberReturned != NULL) {
		*numberReturned = i;
	}

	return 1;
}


int
NWSAPI_GetMeasurements(	const char *seriesName,
			double sinceWhen,
			NWSAPI_Measurement *whereTo,
			unsigned int atMost,
			unsigned int *numberReturned) {
	int ret;
	double lastTimeStamp;
	struct host_cookie memory;
	size_t returnCount;

	/* sanity check */
	if (seriesName == NULL || whereTo == NULL) {
		FAIL("GetMeasurements: NULL parameter(s)\n");
	}
	if (strlen(seriesName) <= 1) {
		FAIL("GetMeasurements: empty seriesName\n");
	}

	/* we try first to use the default memory we have set: if it
	 * doesn't work we go and ask the nameserver */
	returnCount = ret = 0;
	if (defaultMemory.port != 0) {
		if(LoadExperiments(&defaultMemory,
				seriesName,
				whereTo,
				atMost,
				sinceWhen,
				&returnCount,
				&lastTimeStamp,
				-1)) {
			ret = 1;		/* success */
		}
	}

	/* if we didn't get it from the deafult memory, let's go and ask
	 * the nameserver */
	if (ret != 1 && GetSeriesMemory(seriesName, &memory, -1)) {
		if (LoadExperiments(&memory,
				seriesName,
				whereTo,
				atMost,
				sinceWhen,
				&returnCount,
				&lastTimeStamp,
				-1)) {
			ret = 1; 		/* success */
		}
		DisconnectHost(&memory);
	}

	if(numberReturned != NULL) {
		*numberReturned = returnCount;
	}

	/* if we didnt' get result let's tell something */
	if (ret != 1) {
		ERROR1("GetMeasurement: couldn't get experiment for %s\n", seriesName);
	}

	return ret;
}

int
NWSAPI_GetForecasts(	const char *name,
			double sinceWhen,
			NWSAPI_ForecastCollection *whereTo,
			unsigned int atMost,
			unsigned int *num) {
	NWSAPI_Measurement *meas;
	NWSAPI_ForecastState *state;
	unsigned int len;

	/* sanity check */
	if (name == NULL || whereTo == NULL || num == NULL) {
		FAIL("GetForecasts: NULL parameter(s)\n");
	}

	/* let's create room for the measurements */
	meas = (NWSAPI_Measurement *)MALLOC((sizeof(NWSAPI_Measurement) * atMost));
	if (meas == NULL) {
		FAIL("GetForecasts: out of memory\n");
	}

	/* let's compute the forecasts */
	state = NWSAPI_NewForecastState();
	if (state == NULL) {
		FAIL("GetForecasts: out of memory\n");
	}

	/* let's retrieve them */
	if (!NWSAPI_GetMeasurements(name, sinceWhen, meas, atMost, &len)) {
		NWSAPI_FreeForecastState(&state);
		FAIL("GetForecasts: failed to retrieve measurements.\n");
	}


	NWSAPI_UpdateForecastState(state, meas, len, whereTo, len);
	NWSAPI_FreeForecastState(&state);
	*num = len;

	return 1;
}

int
NWSAPI_GetNameServerTimeout(	const NWSAPI_HostSpec *whoToAsk,
				NWSAPI_HostSpec *NS,
				int timeout) {
	HostInfo hostInfo;
	NsInfo nsInfo;
	struct host_cookie tmp;
	int ret;

	/* sanity check */
	if (whoToAsk == NULL || NS == NULL) {
		FAIL("GetNameServer NULL parameter(s)!\n");
	}

	/* Basic approach: send an HOST_GET_NS or HOST_TEST message to
	 * whoToAsk to find out what name server it's using. */
	MakeHostCookie(whoToAsk->machineName, whoToAsk->machinePort, &tmp);
	tmp.sd = NO_SOCKET;

	/* First, try sending a HOST_GET_NS message */
	ret = 0;
	if (HostNameServerInfo(&tmp, &nsInfo, timeout)) {
		if (nsInfo.nsType == NWS_NS) {
			*NS = *NWSAPI_MakeHostSpec(nsInfo.nameServer, 0);
			ret = 1;
		} else {
			ERROR1("GetNameServer: unknown name server type %i\n", nsInfo.nsType);
		}
	} else {
		/* HOST_GET_NS failed; try HOST_TEST */
		if (!GetHostInfo(&tmp, &hostInfo, timeout)) {
			FAIL("GetNameServer: failed to get HOST_TEST\n");
		}
		*NS = *NWSAPI_MakeHostSpec(hostInfo.nameServer, 0);
		ret = 1;
	}
	DisconnectHost(&tmp);

	return ret;
}

int
NWSAPI_GetMemoryTimeout(	const NWSAPI_HostSpec *whoToAsk,
				NWSAPI_HostSpec *memory,
				int timeout) {
	NsInfo nsInfo;
	struct host_cookie tmp;
	int ret;

	/* sanity check */
	if (whoToAsk == NULL || memory == NULL) {
		FAIL("GetMemory: NULL parameter(s)!\n");
	}

	MakeHostCookie(whoToAsk->machineName, whoToAsk->machinePort, &tmp);
	tmp.sd = NO_SOCKET;

	/* try sending a HOST_GET_MEMORY message */
	ret = 0;
	if (HostMemoryInfo(&tmp, &nsInfo, timeout)) {
		if (nsInfo.nsType == NWS_MEMORY) {
			*memory = *NWSAPI_MakeHostSpec(nsInfo.nameServer, 0);
			ret = 1;
		} else {
			ERROR1("GetMemory: unknown memory type %i\n", nsInfo.nsType);
		}
	} else {
		ERROR("GetMemory: couldn't get memory, perhaps old sensor?\n");
	}
	DisconnectHost(&tmp);

	return ret;
}

int
NWSAPI_GetObjectsTimeout(	const char *filter,
				NWSAPI_ObjectSet *whereTo,
				int timeout) {
	int i, ret, len;
	ObjectSet tmp, returnedValue;
	
	/* sanity check */
	if (filter == NULL || whereTo == NULL) {
		FAIL("GetObjects: NULL parameter(s)\n");
	}

	/* let's go through all the nameservers to find what we need */
	ret = 0;

	GetNWSLock(&lock);
	/* let's check if we have NS to talk to */
	if (howManyNS <= 0) {
		WARN("GetObjects: no NameServer has been set!\n");
	}

	/* let's go for it */
	returnedValue = NewObjectSet();
	for (i = 0; i < howManyNS; i++) {
		tmp = NewObjectSet();
		if (!RetrieveObjects(&NSes[i], filter, &tmp, timeout)) {
			ERROR2("GetObjects: unable to retrieve objects from %s (filter '%s')\n", HostCImage(&NSes[i]), filter);
		} else {
			/* we got at least something back */
			ret = 1;

			/* if we got an empty object, let's move on */
			if (strlen(tmp) <= OBJECT_TERMINATOR_LEN) {
				FREE(tmp);
				continue;
			}

			len = strlen(returnedValue);
			returnedValue = REALLOC(returnedValue, len + strlen(tmp) + 1);
			if (returnedValue == NULL) {
				ABORT("GetObjects: out of memory\n");
			}
			/* copy the terminating char too */
			memcpy(returnedValue + len, tmp, strlen(tmp) + 1);
		}
		FREE(tmp);
	}
	ReleaseNWSLock(&lock);

	if (ret == 1) {
		*whereTo = returnedValue;
	} else {
		FREE(returnedValue);
	}

	return ret;
}

int
NWSAPI_IsResourceValid(const char *name) {
	MeasuredResources r;
	int len;

	/* sanity check */
	if (name == NULL) {
		ERROR("IsResourceValid: NULL parameter\n");
		return -1;
	}

	/* ResourceName's return value can be overwritten */
	len = strlen(name);
	GetNWSLock(&lock);
	for(r = 0; r < RESOURCE_COUNT; r++) {
		if (strncmp(ResourceName(r), name, len) == 0){
			/* found it */
			break;
		}
	}
	ReleaseNWSLock(&lock);
	if (r >= RESOURCE_COUNT) {
		ERROR1("IsResourceValid: unknown resource (%s)\n", name);
		return -1;
	}

	return (int) r;
}


int
NWSAPI_IntermachineResource(const char *name) {
	int i, j, len;
	const MeasuredResources *res;

	/* let's get the resource index */
	i = NWSAPI_IsResourceValid(name);
	if (i < 0) {
		FAIL1("IntermachineResource: unknown resource (%s)\n", name);
	}

	/* let's look for the responsible skill */
	for (j = 0; j < SKILL_COUNT; j++) {
		if (!SkillResources((KnownSkills) j, &res, &len)) {
			LOG("IntermachineResource: unknown skill!\n");
			continue;
		} 

		for (; len > 0; len--) {
			if (res[len - 1] != (MeasuredResources) i) {
				/* it's not the right resouce */
				continue;
			}

			/* does the skill do clique measurments? */
			return SkillAvailableForControl((KnownSkills) j, "", CLIQUE_CONTROL);
		}
	}
	/* we shouldn't get here! */
	ERROR("IntermachineResource: run out of skills!\n");

	return 0;
}

const NWSAPI_SeriesSpec *
NWSAPI_MakeSeriesSpec(const char *sourceMachine,
                      const char *destinationMachine,
                      const char *resourceName) {
	static NWSAPI_SeriesSpec returnValue;

	SAFESTRCPY(returnValue.sourceMachine, sourceMachine);
	SAFESTRCPY(returnValue.destinationMachine,
		(destinationMachine == NULL) ? "" : destinationMachine);
	SAFESTRCPY(returnValue.resourceName, resourceName);

	return &returnValue;
}


const NWSAPI_HostSpec *
NWSAPI_MakeHostSpec(const char *host,
                    NWSAPI_Port machinePort) {
	struct host_cookie hostD;
	static NWSAPI_HostSpec returnValue;

	Host2Cookie(host, machinePort, &hostD);
	SAFESTRCPY(returnValue.machineName, hostD.name);
	returnValue.machinePort = hostD.port;

	return &returnValue;
}


NWSAPI_NwsAttribute
NWSAPI_NextNwsAttribute(NWSAPI_Object object,
                     NWSAPI_NwsAttribute current) {
	return NextNwsAttribute(object, (current == NULL) ? NO_ATTRIBUTE : current);
}

NWSAPI_NwsAttribute
NWSAPI_FindNwsAttribute(	const NWSAPI_Object object,
				const char *name) {
	return FindNwsAttribute(object, name);
}


NWSAPI_Object
NWSAPI_NextObject(NWSAPI_ObjectSet objects,
                  NWSAPI_Object current) {
	return NextObject(objects, (current == NULL) ? NO_OBJECT : current);
}


#define KEEP_A_LONG_TIME 315360000.0
int
NWSAPI_PutMeasurements(	const char *seriesName,
			const NWSAPI_SeriesSpec *whichSeries,
			const NWSAPI_Measurement *measurements,
			unsigned int howMany) {
	struct host_cookie localMemory;
	NWSAPI_HostSpec memorySpec;
	struct host_cookie *memoryToUse;
	Object toRegister;
	unsigned int i;
	int ret;
	char *options;

	/* sanity check */
	if (seriesName == NULL || whichSeries == NULL || measurements == NULL) {
		FAIL("PutMeasurements: NULL parameters!\n");
	}

	/* let's try to talk the memory if we have it ... */
	if ((defaultMemory.port != 0) && (defaultMemory.name[0] != '\0')) {
		memoryToUse = &defaultMemory;
	} else {
		/* let's find a memory from a nameserver */
		if (!NWSAPI_GetHosts("hostType=memory", &memorySpec, 1, &i)) {
			i = 0;
		}
		if (i <= 0) {
			FAIL("PutMeasurements: couldn't get a memory!\n");
		}
		SAFESTRCPY(localMemory.name, memorySpec.machineName);
		localMemory.port = memorySpec.machinePort;
		memoryToUse = &localMemory;
	}

	/* let's try to talk to the memory */
	if (!ConnectToHost(memoryToUse, NULL)) {
		FAIL("PutMeasurements: couldn't talk to a memory!\n");
	}

	/* let's generate the series, and the registration */
	options = ResourceOptions(whichSeries);
	toRegister = CreateSeriesObject(seriesName, 
			whichSeries->sourceMachine,
			NULL,
			"noname",
			options,
			whichSeries->resourceName,
			NULL);
	if (toRegister == NULL) {
		FreeObject(&toRegister);
		DisconnectHost(memoryToUse);
		FAIL("PutMeasurements: out of memory\n");
	}

	/* and let's store it. The memory will take care or registering
	 * the series. */
	ret = StoreNewExperiments(	memoryToUse,
					toRegister,
					measurements,
					howMany,
					KEEP_A_LONG_TIME);
	FreeObject(&toRegister);

	return ret;
}


#define MAX_OPTIONS 64
int
NWSAPI_StartActivity(	const NWSAPI_HostSpec *whichSensor,
			const char *activityName,
			const char *controlName,
			const char *skillName,
			const char **options,
			unsigned int howManyOptions) {
	DataDescriptor activityDescriptor = SIMPLE_DATA(CHAR_TYPE, 0);
	char *activitySpec;
	NWSAPI_NwsAttribute controlAttr;
	char controlFilter[127 + 1];
	NWSAPI_Object controlObject;
	NWSAPI_ObjectSet controlObjects;
	int i, j, valueCount, returnValue;
	size_t ignored;
	const char *nameEnd;
	char optionNames[255 + 1];
	OptionInfo optInfo[MAX_OPTIONS];
	unsigned int optInfoCount;
	char portImage[MAX_PORT_IMAGE + 1];
	char sensorName[MAX_MACHINE_NAME + MAX_PORT_IMAGE + 1];
	struct host_cookie sensorLink;
	NWSAPI_NwsAttribute skillAttr;
	char skillFilter[127 + 1];
	NWSAPI_Object skillObject;
	NWSAPI_ObjectSet skillObjects;

	/* Retrieve control and skill registrations from the name server. */
	sprintf(portImage, "%d", whichSensor->machinePort);
	vstrncpy(sensorName, sizeof(sensorName), 3,
		whichSensor->machineName, ":", portImage);
	vstrncpy(controlFilter, sizeof(controlFilter), 12,
		"(&", NWSAPI_CONTROLS,
		"(", NWSAPI_HOST_ATTR, "=", sensorName, ")",
		"(", NWSAPI_CONTROL_ATTR, "=", controlName, "))");
	vstrncpy(skillFilter, sizeof(skillFilter), 12,
		"(&", NWSAPI_SKILLS,
		"(", NWSAPI_HOST_ATTR, "=", sensorName, ")",
		"(", NWSAPI_SKILL_ATTR, "=", skillName, "))");

	if(!NWSAPI_GetObjects(controlFilter, &controlObjects)) {
		FAIL("StartActivity: unable to retrieve sensor control information\n");
	} else if((controlObject = NWSAPI_FirstObject(controlObjects)) == NULL) {
		NWSAPI_FreeObjectSet(&controlObjects);
		FAIL2("StartActivity: no control named %s is registered by %s\n", controlName, sensorName);
	} if(!NWSAPI_GetObjects(skillFilter, &skillObjects)) {
		NWSAPI_FreeObjectSet(&controlObjects);
		FAIL("StartActivity: unable to retrieve sensor skill information\n");
	} else if((skillObject = NWSAPI_FirstObject(skillObjects)) == NULL) {
		NWSAPI_FreeObjectSet(&controlObjects);
		NWSAPI_FreeObjectSet(&skillObjects);
		FAIL2("StartActivity: no skill named %s is registered by %s\n", skillName, sensorName);
	}

	/* Parse each skill option attribute value into an element of
	 * optInfo. */
	controlAttr = FindNwsAttribute(controlObject, NWSAPI_OPTION_ATTR);
	SAFESTRCPY(optionNames, (controlAttr == NULL) ? "" : NwsAttributeValue(controlAttr));
	skillAttr = FindNwsAttribute(skillObject, NWSAPI_OPTION_ATTR);
	if((skillAttr != NULL) && (*NwsAttributeValue(skillAttr) != '\0')) {
		if(optionNames[0] != '\0')
			strcat(optionNames, ",");
		strcat(optionNames, NwsAttributeValue(skillAttr));
	}
	optInfoCount = 0;
	NWSAPI_ForEachNwsAttribute(controlObject, controlAttr) {
		if(ParseOption(controlAttr, optionNames, &optInfo[optInfoCount])) {
			optInfoCount++;
		}
	}
	NWSAPI_ForEachNwsAttribute(skillObject, skillAttr) {
		if(ParseOption(skillAttr, optionNames, &optInfo[optInfoCount])) {
			optInfoCount++;
		}
	}
	NWSAPI_FreeObjectSet(&controlObjects);
	NWSAPI_FreeObjectSet(&skillObjects);

	/* Make sure all names in #options# are valid option names... */
	for(i = 0; i < howManyOptions; i++) {
		nameEnd = strchr(options[i], ':');
		if(nameEnd == NULL) {
			FAIL1("StartActivity: bad option format %s\n", options[i]);
		}
		for(j = 0; j < optInfoCount; j++) {
			if (strncmp(optInfo[j].name, options[i], nameEnd - options[i]) == 0) {
				/* found it */
				break;
			}
		}
		if(j == optInfoCount) {
			FAIL1("StartActivity: unknown option %s\n", options[i]);
		}
	}

	/* ... and that #options# includes a proper number of each option. */
	for(i = 0; i < optInfoCount; i++) {
		valueCount = 0;
		for(j = 0; j < howManyOptions; j++) {
			nameEnd = strchr(options[j], ':');
			if (strncmp(optInfo[i].name, options[j], nameEnd - options[j]) == 0) {
				for(nameEnd++; nameEnd != NULL; nameEnd = strchr(nameEnd + 1, ',')) {
					valueCount++;
				}
			}
		}
		if((valueCount < optInfo[i].minOccurrences) || (valueCount > optInfo[i].maxOccurrences)) {
			FAIL4("StartActivity: %d values given for %s option; %d to %d allowed\n", valueCount, optInfo[i].name, optInfo[i].minOccurrences, optInfo[i].maxOccurrences);
		}
	}

	Host2Cookie(sensorName, DefaultHostPort(SENSOR_HOST), &sensorLink);
	if(!ConnectToHost(&sensorLink, &sensorLink.sd)) {
		FAIL1("StartActivity: unable to contact sensor %s\n", sensorName);
	}

	/* Package the registration name, control name, skill name, and
	 * options into a series of nul-terminated strings to accompany
	 * the start message.
	 */
	activityDescriptor.repetitions = strlen(activityName) + strlen(controlName) + strlen(skillName) + 4;
	for(i = 0; i < howManyOptions; i++) {
		activityDescriptor.repetitions += strlen(options[i]) + 1;
	}
	activitySpec = (char *)MALLOC(activityDescriptor.repetitions);
	if(activitySpec == NULL) {
		DisconnectHost(&sensorLink);
		FAIL("StartActivity: out of memory\n");
	}

	/* Use spaces to reserve space for the nul characters. */
	vstrncpy(activitySpec, activityDescriptor.repetitions, 6,
		activityName, " ", controlName, " ", skillName, " ");
	for(i = 0; i < howManyOptions; i++) {
		strcat(activitySpec, options[i]);
		strcat(activitySpec, "\t");
	}
	activitySpec[strlen(activityName)] = '\0';
	activitySpec[strlen(activityName) + strlen(controlName) + 1] = '\0';
	activitySpec[strlen(activityName) + strlen(controlName) + strlen(skillName) + 2] = '\0';

	/* Finally, send the start message. */
	returnValue = SendMessageAndData(sensorLink.sd,
			ACTIVITY_START,
			activitySpec,
			&activityDescriptor,
			1,
			-1) &&
		RecvMessage(sensorLink.sd,
			ACTIVITY_STARTED,
			&ignored,
			-1);
	DisconnectHost(&sensorLink);
	FREE(activitySpec);

	return returnValue;
}

int
NWSAPI_RestartActivities(const NWSAPI_HostSpec *whichSensor) {

  char portImage[MAX_PORT_IMAGE + 1];
  char sensorName[MAX_MACHINE_NAME + MAX_PORT_IMAGE + 1];
  char controlFilter[127 + 1];
  NWSAPI_NwsAttribute controlAttr;
  NWSAPI_Object controlObject;
  NWSAPI_ObjectSet controlObjectSet;
  char *optionstring,*word;
  int i;
  
  /* args for StartActivity */
  char activityName[127 + 1];
  char controlName[127 + 1];
  char skillName[127 + 1];
  char **options=NULL;
  unsigned int howManyOptions=0;

  /* Retrieve control and skill registrations from the name server. */
  sprintf(portImage, "%d", whichSensor->machinePort);
  vstrncpy(sensorName, sizeof(sensorName), 3,
           whichSensor->machineName, ":", portImage);
  vstrncpy(controlFilter, sizeof(controlFilter), 7,
           "(&", NWSAPI_ACTIVITIES,
          "(", NWSAPI_HOST_ATTR, "=", sensorName, "))");

  LOG1("Restart all activities on sensor %s\n",sensorName);

  if(!NWSAPI_GetObjects(controlFilter, &controlObjectSet)) {
    FAIL("RestartActivities: unable to retrieve activity informations\n");
  }
  else if(!NWSAPI_FirstObject(controlObjectSet)) {
    NWSAPI_FreeObjectSet(&controlObjectSet);
    FAIL1("RestartActivities: no activities is registered by %s\n",
         sensorName);
  }
  NWSAPI_ForEachObject(controlObjectSet, controlObject) {
    activityName[0]='\0';
    controlName[0]='\0';
    for (i=0;i<howManyOptions;i++)
      FREE(options[i]);
    FREE(options); 
    options=NULL;
    howManyOptions = 0;

    NWSAPI_ForEachNwsAttribute(controlObject, controlAttr) {
      if(!strcmp(NWSAPI_NwsAttributeName(controlAttr), NWSAPI_NAME_ATTR)) {
       vstrncpy(activityName,128,1,NWSAPI_NwsAttributeValue(controlAttr));

      } else if (!strcmp(NWSAPI_NwsAttributeName(controlAttr),
                        NWSAPI_CONTROL_ATTR)) {
       vstrncpy(controlName,128,1,NWSAPI_NwsAttributeValue(controlAttr));

      } else if (!strcmp(NWSAPI_NwsAttributeName(controlAttr),
                        NWSAPI_SKILL_ATTR)) {
       vstrncpy(skillName,128,1,NWSAPI_NwsAttributeValue(controlAttr));

      } else if (!strcmp(NWSAPI_NwsAttributeName(controlAttr),
                        NWSAPI_OPTION_ATTR)) {
       optionstring=strdup(NWSAPI_NwsAttributeValue(controlAttr));
       for(word = strtok(optionstring, ",");
           word != NULL;
           word = strtok(NULL, ",")) {
         howManyOptions++;
         if (!(options=REALLOC(options,howManyOptions))
             || !(options[howManyOptions-1]=(char*)MALLOC(sizeof(char)*128))){
            FAIL("Out of memory\n"); 
         }
         snprintf(options[howManyOptions-1],128,word);
       }
       FREE(optionstring);
       
      } else { /* search for an option */
       for (i=0 ; i<howManyOptions ; i++) {
         if (!strcmp(NWSAPI_NwsAttributeName(controlAttr),options[i])) {
           strcat(options[i],":");
           strcat(options[i],NWSAPI_NwsAttributeValue(controlAttr));
         }
       }
      }
    }

    if (!NWSAPI_StartActivity(whichSensor,activityName,controlName,skillName,
                             (const char**)options,howManyOptions)) {
      NWSAPI_FreeObjectSet(&controlObjectSet);
      for (i=0;i<howManyOptions;i++)
       FREE(options[i]);
      FREE(options);
      return 0;
    }
  }

  for (i=0;i<howManyOptions;i++)
    FREE(options[i]);
  FREE(options);
  NWSAPI_FreeObjectSet(&controlObjectSet);
  return 1;
}



char *
NWSAPI_SeriesName_r(const NWSAPI_SeriesSpec *whichSeries) {
	char *options, *tmp;

	options = ResourceOptions(whichSeries);
	if (options == NULL) {
		ERROR("SeriesName_r: bad SeriesSpec\n");
		return NULL;
	}

	tmp = NameOfSeries_r(	whichSeries->sourceMachine, 
				whichSeries->resourceName,
				options);
	FREE(options);

	return (tmp);
}


const char *
NWSAPI_SeriesName(const NWSAPI_SeriesSpec *whichSeries) {
	char *tmp; 
	static char returnValue[127 + 1];

	returnValue[0] = '\0';

	tmp =  NWSAPI_SeriesName_r(whichSeries);
	if (tmp != NULL) {
		SAFESTRCPY(returnValue, tmp);
	}
	FREE(tmp);

	return &returnValue[0];
}


int
NWSAPI_UseMemory(const NWSAPI_HostSpec *whichMemory) {
	return UseHost(whichMemory, &defaultMemory);
}


int
NWSAPI_UseNameServer(const NWSAPI_HostSpec *whichNS) {
	struct host_cookie tmp;
	int i;

	/* if NULL, we need to reset the list of namesevers */
	if (whichNS == NULL || strlen(whichNS->machineName) <= 0) {
		GetNWSLock(&lock);
		/* let's drop the sockets */
		for (i = 0; i < howManyNS; i++) {
			DisconnectHost(&NSes[i]);
		}

		/* and clean up the memory */
		FREE(NSes);
		howManyNS = 0;
		ReleaseNWSLock(&lock);
	} else {
		/* let's add the nameserver to our list */
		if (!UseHost(whichNS, &tmp)) {
			/* it was no good */
			return 0;
		}

		/* we need to lock */
		GetNWSLock(&lock);
		NSes = REALLOC(NSes, sizeof(struct host_cookie) * (howManyNS + 1));
		if (NSes == NULL) {
			ReleaseNWSLock(&lock);
			FAIL("UseNameServer: out of memory\n");
		}
		NSes[howManyNS] = tmp;
		howManyNS++;
		ReleaseNWSLock(&lock);
	}

	return 1;
}

int
NWSAPI_GetAllForecasts(	NWSAPI_HostSpec *proxy,
			char *resource,
			char *opts,
			char **hosts,
			NWSAPI_ForecastCollection **col,
			int timeOut) {
	struct host_cookie tmp;
	char *request;
	int i, howMany;
	NWSAPI_ForecastCollection *c;
	DataDescriptor des = SIMPLE_DATA(CHAR_TYPE, 0);
	DataDescriptor howManyDes = SIMPLE_DATA(INT_TYPE, 1);

	/* sanity check */
	if (proxy == NULL || resource == NULL || hosts == NULL || col == NULL) {
		ERROR("GetAllForecasts: NULL parameter!\n");
		return 0;
	}

	/* let's connect to the proxy */
	MakeHostCookie(proxy->machineName, proxy->machinePort, &tmp);
	if (!ConnectToHost(&tmp, NULL)) {
		FAIL1("GetAllForecasts: failed to connect to %s\n", HostCImage(&tmp));
	}

	/* let's create the request */
	des.repetitions = strlen(resource) + 1;
	if (opts != NULL) {
		des.repetitions += strlen(opts) + 1;
	} else {
		des.repetitions += 1;
	}
	for (i = 0; hosts[i] != NULL; i++) {
		des.repetitions += strlen(hosts[i]) + 1;
	}
	request = MALLOC(sizeof(char) * des.repetitions);
	if (request == NULL) {
		DROP_SOCKET(&tmp.sd);
		FAIL("GetAllForecasts: out of memory\n");
	}
	memcpy(request, resource, strlen(resource) + 1);
	des.repetitions = strlen(resource) + 1;
	if (opts != NULL) {
		memcpy(request + des.repetitions, opts, strlen(opts) + 1);
		des.repetitions += strlen(opts) + 1;
	} else {
		/* empty string */
		memcpy(request + des.repetitions, "\0", 1);
		des.repetitions += 1;
	}
	for (i = 0; hosts[i] != NULL; i++) {
		memcpy(request + des.repetitions, hosts[i], strlen(hosts[i])+1);
		des.repetitions += strlen(hosts[i]) + 1;
	}
	howMany = i--;

	/* let's send the request */
	if (!SendMessageAndData(tmp.sd, GET_FORECASTS, request, &des, 1, timeOut)) {
		FREE(request);
		DROP_SOCKET(&tmp.sd);
		FAIL1("GetAllForecasts: failed to talk %s\n", HostCImage(&tmp));
	}
	FREE(request);

	/* let's get the forecasts */
	if (!RecvMessageAndData(tmp.sd, GOT_FORECASTS, &i, &howManyDes, 1, timeOut)) {
		DROP_SOCKET(&tmp.sd);
		FAIL1("GetAllForecasts: failed to talk %s\n", HostCImage(&tmp));
	}
	if ((howMany*howMany) < i) {
		FAIL("GetAllForecasts: too many forecasts!\n");
	}

	/* allocate space */
	c = (NWSAPI_ForecastCollection *)MALLOC(sizeof(NWSAPI_ForecastCollection)*i);
	if (c == NULL) {
		FAIL("GetAllForecasts: out of memory\n");
	}

	/* we receive the forecastCollection one at a time */
	howMany = i;
	for (i = 0; i < howMany; i++) {
		if (!RecvData(tmp.sd, &c[i], forecastCollectionDescriptor, forecastCollectionDescriptorLength, timeOut)) {
			FREE(c);
			FAIL("GetAllForecasts: failed to receive forecasts\n");
		}
	}
	DROP_SOCKET(&tmp.sd);
	*col = c;

	return howMany;
}

int
NWSAPI_StartFetching(	NWSAPI_HostSpec *proxy,
			char *resource,
			char *opts,
			char **hosts,
			int timeOut) {
	struct host_cookie tmp;
	char *request;
	int i;
	DataDescriptor des = SIMPLE_DATA(CHAR_TYPE, 0);

	/* sanity check */
	if (proxy == NULL || resource == NULL || hosts == NULL) {
		ERROR("StartFetching: NULL parameter!\n");
		return 0;
	}

	/* let's connect to the proxy */
	MakeHostCookie(proxy->machineName, proxy->machinePort, &tmp);
	if (!ConnectToHost(&tmp, NULL)) {
		FAIL1("StartFetching: failed to connect to %s\n", HostCImage(&tmp));
	}

	/* let's create the request */
	des.repetitions = strlen(resource) + 1;
	if (opts != NULL) {
		des.repetitions += strlen(opts) + 1;
	} else {
		des.repetitions += 1;
	}
	for (i = 0; hosts[i] != NULL; i++) {
		des.repetitions += strlen(hosts[i]) + 1;
	}
	request = MALLOC(sizeof(char) * des.repetitions);
	if (request == NULL) {
		DROP_SOCKET(&tmp.sd);
		FAIL("StartFetching: out of memory\n");
	}
	memcpy(request, resource, strlen(resource) + 1);
	des.repetitions = strlen(resource) + 1;
	if (opts != NULL) {
		memcpy(request + des.repetitions, opts, strlen(opts) + 1);
		des.repetitions += strlen(opts) + 1;
	} else {
		/* empty string */
		memcpy(request + des.repetitions, "\0", 1);
		des.repetitions += 1;
	}
	for (i = 0; hosts[i] != NULL; i++) {
		memcpy(request + des.repetitions, hosts[i], strlen(hosts[i])+1);
		des.repetitions += strlen(hosts[i]) + 1;
	}

	/* let's send the request */
	if (!SendMessageAndData(tmp.sd, START_FETCHING, request, &des, 1, timeOut)) {
		FREE(request);
		DROP_SOCKET(&tmp.sd);
		FAIL1("StartFetching: failed to talk %s\n", HostCImage(&tmp));
	}
	FREE(request);

	/* let's get the forecasts */
	if (!RecvMessage(tmp.sd, PROXY_ACK, (size_t *)&i, timeOut)) {
		DROP_SOCKET(&tmp.sd);
		ERROR("StartFetching: failed to receive ack\n");
	}
	DROP_SOCKET(&tmp.sd);

	return 1;
}