/*
 * scamper_control
 *
 * $Id: scamper_control.c,v 1.85 2007/05/13 23:35:48 mjl Exp $
 *
 * if scamper is started as a daemon that listens for commands, then this
 * file contains the logic that drives the daemon.
 * 
 * by default, scamper listens locally for commands.  there are plans to
 * status notifications out to interested parties for the purpose of scamper
 * process monitoring.
 *
 * Copyright (C) 2004-2007 The University of Waikato
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, version 2.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 *
 */

#if defined(__APPLE__)
#define _BSD_SOCKLEN_T_
#include <stdint.h>
#endif

#include <sys/types.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <netinet/in.h>

#include <errno.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include <stdio.h>
#include <ctype.h>
#include <stdarg.h>

#include <assert.h>

#if defined(DMALLOC)
#include <dmalloc.h>
#endif

#include "scamper.h"
#include "scamper_addr.h"
#include "scamper_list.h"
#include "scamper_tlv.h"
#include "scamper_trace.h"
#include "scamper_control.h"
#include "scamper_debug.h"
#include "scamper_fds.h"
#include "scamper_linepoll.h"
#include "scamper_writebuf.h"
#include "scamper_file.h"
#include "scamper_outfiles.h"
#include "scamper_task.h"
#include "scamper_queue.h"
#include "scamper_addresslist.h"
#include "mjl_list.h"
#include "utils.h"

typedef struct client
{
  struct sockaddr    *sa;
  scamper_fd_t       *fdn;
  scamper_linepoll_t *lp;
  scamper_writebuf_t *wb;
  void               *observe;
  dlist_node_t       *node;
} client_t;

typedef struct command
{
  char *word;
  int (*handler)(client_t *client, char *param);
} command_t;

typedef struct param
{
  char  *word;
  char **var;
} param_t;

/*
 * client_list: a doubly linked list of connected clients
 * fd: a scamper_fd struct that contains callback details
 */
static dlist_t      *client_list  = NULL;
static scamper_fd_t *fdn          = NULL;

static int command_handler(command_t *handler, int cnt, client_t *client,
			   char *word, char *param, int *retval)
{
  int i;

  for(i=0; i<cnt; i++)
    {
      if(strcasecmp(handler[i].word, word) == 0)
	{
	  *retval = handler[i].handler(client, param);
	  return 0;
	}
    }

  return -1;
}

/*
 * params_get
 *
 * go through the line and get parameters out, returning the start of
 * each parameter in the words array.
 */
static int params_get(char *line, char **words, int *count)
{
  int i, w;

  i = 0; /* first character in the parameters */
  w = 0; /* first word to be read */

  /* if there is no line, there can't be any parameters */
  if(line == NULL)
    {
      *count = 0;
      return 0;
    }

  while(line[i] != '\0' && w < *count)
    {
      if(line[i] == '"')
	{
	  /* the start of the parameter is past the opening quote */
	  words[w++] = &line[++i];

	  /* until we get to the end of the param / string, keep hunting */
	  while(line[i] != '"' && line[i] != '\0') i++;

	  /* did not get the closing double-quote */
	  if(line[i] == '\0') return -1;
	}
      else
	{
	  /* the start of the word is here, skip past this opening char */
	  words[w++] = &line[i++];

	  /* until we get to the end of the word / string, keep hunting */
	  while(line[i] != ' ' && line[i] != '\0') i++;

	  if(line[i] == '\0') break;

	}

      /* null terminate the word, skip towards the next word */
      line[i++] = '\0';

      /* skip to the next word */
      while(line[i] == ' ' && line[i] != '\0') i++;
    }

  if(line[i] == '\0')
    {
      *count = w;
      return 0;
    }

  return -1;
}

static char *switch_tostr(char *buf, size_t len, int val)
{
  if(val == 0)
    {
      strncpy(buf, "off", len);
    }
  else
    {
      strncpy(buf, "on", len);
    }

  return buf;
}

/*
 * client_free
 *
 * free up client state for the socket handle.
 */
static void client_free(client_t *client)
{
  int fd;

  if(client == NULL) return;

  /* if there's an open socket here, close it now */
  if(client->fdn != NULL)
    {
      fd = scamper_fd_fd_get(client->fdn);
      scamper_fd_free(client->fdn);

      shutdown(fd, SHUT_RDWR);
      close(fd);
    }

  /* remove the linepoll structure */
  if(client->lp != NULL) scamper_linepoll_free(client->lp, 0);

  /* remove the writebuf structure */
  if(client->wb != NULL) scamper_writebuf_free(client->wb);

  /* remove the client from the list of clients */
  if(client->node != NULL) dlist_node_pop(client_list, client->node);

  /* if we made a copy of the client's sockaddr, free it now */
  if(client->sa != NULL) free(client->sa);

  /* if we are monitoring source events, unobserve */
  if(client->observe != NULL) scamper_addresslist_unobserve(client->observe);

  free(client);
  return;
}

static int client_send(client_t *client, char *fs, ...)
{
  char    msg[512], *str;
  va_list ap;
  int     ret;
  size_t  len;

  va_start(ap, fs);
  if((ret = vsnprintf(msg, sizeof(msg), fs, ap)) > (int)sizeof(msg))
    {
      len = ret;

      if((str = malloc((size_t)(len + 2))) == NULL)
	{
	  va_end(ap);
	  return -1;
	}
      vsnprintf(str, len+1, fs, ap);
      va_end(ap);

      str[len++] = '\n'; 
      str[len] = '\0';

      ret = scamper_writebuf_send(client->wb, str, len);
      free(str);
    }
  else
    {
      len = ret;

      va_end(ap);

      msg[len++] = '\n'; 
      msg[len] = '\0';

      ret = scamper_writebuf_send(client->wb, msg, len);
    }

  return ret;
}

/*
 * param_handler
 *
 */
static int param_handler(param_t *handler, int cnt, client_t *client,
			 char *param, char *next)
{
  int i;

  for(i=0; i<cnt; i++)
    {
      /* skip until we find the handler for this parameter */
      if(strcasecmp(handler[i].word, param) != 0)
	{
	  continue;
	}

      /* already seen this parameter specified */
      if(*handler[i].var != NULL)
	{
	  client_send(client, "ERR parameter '%s' already specified", param);
	  return -1;
	}

      /* the parameter passed does not have a value to go with it */
      if(next == NULL)
	{
	  client_send(client, "ERR parameter '%s' requires argument", param);
	  return -1;
	}

      /* got the parameter */
      *handler[i].var = next;
      return 0;
    }

  return -1;
}

static int set_long(client_t *client, char *buf, char *name,
		    int (*setfunc)(int), int min, int max)
{
  long l;

  if(buf == NULL)
    {
      client_send(client, "ERR set %s requires argument", name);
      return -1;
    }

  /*
   * null terminate this word.  discard the return value, we don't care
   * about any further words.
   */
  string_nextword(buf);

  /* make sure the argument is an integer argument */
  if(string_isnumber(buf) == 0)
    {
      client_send(client, "ERR set %s argument is not an integer", name);
      return -1;
    }

  /* convert the argument to a long.  catch any error */
  if(string_tolong(buf, &l) != 0)
    {
      client_send(client, "ERR %s", strerror(errno));
      return -1;
    }

  if(setfunc(l) == -1)
    {
      client_send(client, "ERR %s of range (%d, %d)", name, min, max);
      return -1;
    }

  client_send(client, "OK %s %d", name, l);
  return 0;
}

static int get_switch(client_t *client, char *name, char *buf, long *l)
{
  if(strcasecmp(buf, "on") == 0)
    {
      *l = 1;
    }
  else if(strcasecmp(buf, "off") == 0)
    {
      *l = 0;
    }
  else
    {
      client_send(client, "ERR %s <on|off>", name);
      return -1;
    }

  return 0;
}

static char *source_tostr(char *str, const size_t len,
			  const scamper_source_t *source)
{
  char descr[256], file[256], autoreload[4], adhoc[4], outfile[256];

  /* if there is a description for the source, then format it in */
  if(scamper_source_getdescr(source) != NULL)
    {
      snprintf(descr, sizeof(descr),
	       " descr '%s'", scamper_source_getdescr(source));
    }
  else descr[0] = '\0';

  /*
   * if there is a filename associated with the source, then format it in;
   * also format the cycle details, and autoreload details, since that only
   * makes sense when a file is used to feed addresses in.
   */
  if(scamper_source_getfilename(source) != NULL)
    {
      snprintf(file, sizeof(file),
	       " file '%s' cycle_id %d cycles %d autoreload %s",
	       scamper_source_getfilename(source),
	       scamper_source_getcycleid(source),
	       scamper_source_getcycles(source),
	       switch_tostr(autoreload, sizeof(autoreload),
			    scamper_source_getautoreload(source)));
    }
  else file[0] = '\0';

  /* decode the adhoc switch */
  switch_tostr(adhoc, sizeof(adhoc), scamper_source_getadhoc(source));

  /* outfile */
  if(scamper_source_getoutfile(source) != NULL)
    {
      snprintf(outfile, sizeof(outfile), " outfile '%s'",
	      scamper_outfile_getname(scamper_source_getoutfile(source)));
    }
  else outfile[0] = '\0';

  snprintf(str, len,
	   "name '%s'%s list_id %d%s priority %d adhoc %s%s",
	   scamper_source_getname(source),
	   descr,
	   scamper_source_getlistid(source),
	   file,
	   scamper_source_getpriority(source),
	   adhoc,
	   outfile);

  return str;
}

static int command_do(client_t *client, char *buf)
{
  if(buf == NULL)
    {
      client_send(client, "ERR usage: do [ ping | trace ]");
      return 0;
    }

  if(scamper_source_do(NULL, buf) == -1)
    {
      client_send(client, "ERR could not do '%s'", buf);
      return 0;
    }

  client_send(client, "OK");

  return 0;
}

static int command_exit(client_t *client, char *buf)
{
  client_free(client);
  return 0;
}

static int command_get_command(client_t *client, char *buf)
{
  const char *command = scamper_command_get();
  if(command == NULL)
    {
      return client_send(client, "OK null command");
    }
  return client_send(client, "OK command %s", command);
}

static int command_get_holdtime(client_t *client, char *buf)
{
  int holdtime = scamper_holdtime_get();
  return client_send(client, "OK holdtime %d", holdtime);
}

static int command_get_monitorname(client_t *client, char *buf)
{
  const char *monitorname = scamper_monitorname_get();
  if(monitorname == NULL)
    {
      return client_send(client, "OK null monitorname");
    }
  return client_send(client, "OK monitorname %s", monitorname);
}

static int command_get_pid(client_t *client, char *buf)
{
  pid_t pid = getpid();
  return client_send(client, "OK pid %d", pid);
}

static int command_get_pps(client_t *client, char *buf)
{
  int pps = scamper_pps_get();
  return client_send(client, "OK pps %d", pps);
}

static int command_get_sport(client_t *client, char *buf)
{
  int sport = scamper_sport_get();
  return client_send(client, "OK sport %d", sport);
}

static int command_get_version(client_t *client, char *buf)
{
  return client_send(client, "OK version " SCAMPER_VERSION);
}

static int command_get(client_t *client, char *buf)
{
  static command_t handlers[] = {
    {"command",     command_get_command},
    {"holdtime",    command_get_holdtime},
    {"monitorname", command_get_monitorname},
    {"pid",         command_get_pid},
    {"pps",         command_get_pps},
    {"sport",       command_get_sport},
    {"version",     command_get_version},
  };
  static int handler_cnt = sizeof(handlers) / sizeof(command_t);
  int ret;

  if(buf == NULL)
    {
      client_send(client, "ERR usage: get "
	  "[command | holdtime | monitorname | pid | pps | sport | version]");
      return 0;
    }

  if(command_handler(handlers, handler_cnt, client, buf, NULL, &ret) == -1)
    {
      client_send(client, "ERR unhandled get command '%s'", buf);
      return 0;
    }

  return 0;
}

static int command_help(client_t *client, char *buf)
{
  client_send(client, "ERR XXX: todo");
  return 0;
}

static void observe_source_event_add(const scamper_source_event_t *sse,
				     char *buf, const size_t len)
{
  buf[0] = 'a'; buf[1] = 'd'; buf[2] = 'd'; buf[3] = ' ';
  source_tostr(buf+4, len-4, sse->source);
  return;
}

static void observe_source_event_update(const scamper_source_event_t *sse,
					char *buf, const size_t len)
{
  char autoreload[16];
  char cycles[16];
  char priority[24];

  /* autoreload */
  if(sse->sse_update_flags & 0x01)
    snprintf(autoreload, sizeof(autoreload),
	     " autoreload %d", sse->sse_update_autoreload);
  else autoreload[0] = '\0';

  /* cycles */
  if(sse->sse_update_flags & 0x02)
    snprintf(cycles, sizeof(cycles),
	     " cycles %d", sse->sse_update_cycles);
  else cycles[0] = '\0';

  /* priority */
  if(sse->sse_update_flags & 0x04)
    snprintf(priority, sizeof(priority),
	     " priority %d", sse->sse_update_priority);
  else priority[0] = '\0';

  snprintf(buf, len, "update '%s'%s%s%s",
	   scamper_source_getname(sse->source),
	   autoreload, cycles, priority);
  return;
}

static void observe_source_event_cycle(const scamper_source_event_t *sse,
				       char *buf, const size_t len)
{
  snprintf(buf, len, "cycle '%s' id %d",
	   scamper_source_getname(sse->source),
	   sse->sse_cycle_cycle_id);
  return;
}

static void observe_source_event_delete(const scamper_source_event_t *sse,
					char *buf, const size_t len)
{
  snprintf(buf, len, "delete '%s'",
	   scamper_source_getname(sse->source));
  return;
}

static void observe_source_event_finish(const scamper_source_event_t *sse,
					char *buf, const size_t len)
{
  snprintf(buf, len, "finish '%s'",
	   scamper_source_getname(sse->source));
  return;
}

/*
 * command_observe_source_cb
 *
 * this function is a callback that is used whenever some event occurs
 * with a source.
 */
static void command_observe_source_cb(const scamper_source_event_t *sse,
				      void *param)
{
  static void (* const func[])(const scamper_source_event_t *,
			       char *, const size_t) = 
  {
    observe_source_event_add,
    observe_source_event_update,
    observe_source_event_cycle,
    observe_source_event_delete,
    observe_source_event_finish,
  };
  client_t *client = (client_t *)param;
  char buf[512];
  size_t len;

  if(sse->event < 0x01 || sse->event > 0x05)
    {
      return;
    }

  snprintf(buf, sizeof(buf), "EVENT %d source ", (uint32_t)sse->sec);
  len = strlen(buf);

  func[sse->event-1](sse, buf + len, sizeof(buf)-len);
  client_send(client, "%s", buf);

  return;
}

static int command_observe(client_t *client, char *buf)
{
  if(buf == NULL)
    {
      client_send(client, "ERR usage: observe [sources]");
      return 0;
    }
  string_nextword(buf);

  if(strcasecmp(buf, "sources") != 0)
    {
      client_send(client, "ERR usage: observe [sources]");
      return 0;
    }

  client->observe = scamper_addresslist_observe(command_observe_source_cb,
						client);
  if(client->observe == NULL)
    {
      client_send(client, "ERR could not observe");
      return -1;
    }

  client_send(client, "OK");
  return 0;
}

/*
 * command_outfile_close
 *
 * outfile close <alias>
 */
static int command_outfile_close(client_t *client, char *buf)
{
  scamper_outfile_t *sof;

  if(buf == NULL)
    {
      client_send(client, "ERR usage: outfile close <alias>");
      return -1;
    }
  string_nextword(buf);

  if((sof = scamper_outfiles_get(buf)) == NULL)
    {
      client_send(client, "ERR unknown outfile '%s'", buf);
      return -1;
    }

  if(scamper_outfile_close(sof) == -1)
    {
      client_send(client, "ERR could not drop outfile: refcnt %d",
		  scamper_outfile_getrefcnt(sof));
      return -1;
    }

  client_send(client, "OK");
  return 0;
}

static int outfile_foreach(void *param, scamper_outfile_t *sof)
{
  client_t *client = (client_t *)param;
  scamper_file_t *sf = scamper_outfile_getfile(sof);
  char *filename = scamper_file_getfilename(sf);

  if(filename == NULL) filename = "(null)";

  client_send(client, "INFO '%s' file '%s' refcnt %d",
	      scamper_outfile_getname(sof),
	      filename,
	      scamper_outfile_getrefcnt(sof));
  return 0;
}

/*
 * command_outfile_list
 *
 * outfile list
 */
static int command_outfile_list(client_t *client, char *buf)
{
  scamper_outfiles_foreach(client, outfile_foreach);
  client_send(client, "OK");
  return 0;
}

/*
 * command_outfile_open
 *
 * outfile open name <alias> mode <truncate|append> file <path>
 */
static int command_outfile_open(client_t *client, char *buf)
{
  scamper_outfile_t *sof;
  char *params[24];
  int   i, cnt = sizeof(params) / sizeof(char *);
  char *file = NULL, *mode = NULL, *name = NULL;
  char *next;
  param_t handlers[] = {
    {"file", &file},
    {"mode", &mode},
    {"name", &name},
  };
  int handler_cnt = sizeof(handlers) / sizeof(param_t);

  if(params_get(buf, params, &cnt) == -1)
    {
      client_send(client, "ERR params_get failed");
      return -1;
    }

  for(i=0; i<cnt; i += 2)
    {
      if(i+1 != cnt) next = params[i+1];
      else next = NULL;

      if(param_handler(handlers, handler_cnt, client, params[i], next) == -1)
	{
	  client_send(client, "ERR param '%s' failed", params[i]);
	  return -1;
	}
    }

  if(name == NULL || file == NULL || mode == NULL)
    {
      client_send(client,
		  "ERR usage: outfile open name <alias> file <path> "
		  "mode <truncate|append>");
      return -1;
    }

  if(strcasecmp(mode, "truncate") != 0 && strcasecmp(mode, "append") != 0)
    {
      client_send(client, "ERR mode must be truncate or append");
      return -1;
    }

  if((sof = scamper_outfile_open(name, file, mode)) == NULL)
    {
      client_send(client, "ERR could not add outfile");
      return -1;
    }

  client_send(client, "OK");
  return 0;
}

/*
 * outfile socket
 *
 * outfile socket name <alias> type <type>
 */
static int command_outfile_socket(client_t *client, char *buf)
{
  scamper_outfile_t *sof;
  char *params[4], *next;
  int   i, fd;
  int   cnt = sizeof(params) / sizeof(char *);
  char *name = NULL, *type = NULL;
  param_t handlers[] = {
    {"name", &name},
    {"type", &type},
  };
  int handler_cnt = sizeof(handlers) / sizeof(param_t);

  if(params_get(buf, params, &cnt) == -1)
    {
      client_send(client, "ERR source add params_get failed");
      return -1;
    }

  for(i=0; i<cnt; i += 2)
    {
      if(i+1 != cnt) next = params[i+1];
      else next = NULL;

      if(param_handler(handlers, handler_cnt, client, params[i], next) == -1)
	{
	  client_send(client, "ERR source add param '%s' failed", params[i]);
	  return -1;
	}
    }

  if(name == NULL || type == NULL)
    {
      client_send(client, "ERR usage outfile socket name <alias> type <type>");
      return 0;
    }

  if((sof = scamper_outfiles_get(name)) != NULL)
    {
      client_send(client, "ERR outfile '%s' already exists", name);
      return 0;
    }

  fd = scamper_fd_fd_get(client->fdn);
  if((sof = scamper_outfile_openfd(name, fd, type)) == NULL)
    {
      client_send(client, "ERR could not turn socket into outfile");
      return 0;
    }

  client_send(client, "OK");
  return 0;
}

/*
 * outfile swap
 *
 * swap <alias 1> <alias 2>
 */
static int command_outfile_swap(client_t *client, char *buf)
{
  scamper_outfile_t *a, *b;
  char *files[2];
  int   cnt = 2;

  if(params_get(buf, files, &cnt) == -1)
    {
      client_send(client, "ERR params_get failed");
      return -1;
    }

  if(cnt != 2)
    {
      client_send(client, "ERR usage outfile swap <alias 1> <alias 2>");
      return -1;
    }

  if((a = scamper_outfiles_get(files[0])) == NULL)
    {
      client_send(client, "ERR unknown outfile '%s'", a);
      return -1;
    }

  if((b = scamper_outfiles_get(files[1])) == NULL)
    {
      client_send(client, "ERR unknown outfile '%s'", b);
      return -1;
    }

  scamper_outfiles_swap(a, b);
  client_send(client, "OK");

  return 0;
}

static int command_outfile(client_t *client, char *buf)
{
  static command_t handlers[] = {
    {"close",  command_outfile_close},
    {"list",   command_outfile_list},
    {"open",   command_outfile_open},
    {"socket", command_outfile_socket},
    {"swap",   command_outfile_swap},
  };
  static int handler_cnt = sizeof(handlers) / sizeof(command_t);
  char *next;
  int ret;

  if(buf == NULL)
    {
      client_send(client, "ERR usage: outfile [close | list | open | swap]");
      return 0;
    }
  next = string_nextword(buf);

  if(command_handler(handlers, handler_cnt, client, buf, next, &ret) == -1)
    {
      client_send(client, "ERR unhandled outfile command '%s'", buf);
    }

  return 0;
}

static int command_set_command(client_t *client, char *buf)
{
  if(scamper_command_set(buf) == -1)
    {
      client_send(client, "ERR could not set command");
      return -1;
    }

  client_send(client, "OK");
  return 0;
}

static int command_set_holdtime(client_t *client, char *buf)
{
  return set_long(client, buf, "holdtime", scamper_holdtime_set,
		  SCAMPER_HOLDTIME_MIN, SCAMPER_HOLDTIME_MAX);
}

static int command_set_monitorname(client_t *client, char *buf)
{
  if(scamper_monitorname_set(buf) == -1)
    {
      client_send(client, "ERR could not set monitorname");
      return -1;
    }

  client_send(client, "OK");
  return 0;
}

static int command_set_pps(client_t *client, char *buf)
{
  return set_long(client, buf, "pps", scamper_pps_set,
		  SCAMPER_PPS_MIN, SCAMPER_PPS_MAX);
}

static int command_set(client_t *client, char *buf)
{
  static command_t handlers[] = {
    {"command",     command_set_command},
    {"holdtime",    command_set_holdtime},
    {"monitorname", command_set_monitorname},
    {"pps",         command_set_pps},
  };
  static int handler_cnt = sizeof(handlers) / sizeof(command_t);
  char *next;
  int ret;

  if(buf == NULL)
    {
      client_send(client,
		  "ERR usage: set [command | holdtime | monitorname | pps]");
      return 0;
    }
  next = string_nextword(buf);

  if(command_handler(handlers, handler_cnt, client, buf, next, &ret) == -1)
    {
      client_send(client, "ERR unhandled set command '%s'", buf);
    }
  return 0;
}

/*
 * command_source_add
 *
 * source add [name <name>] [descr <descr>] [command <command>]
 *            [list_id <id>] [cycle_id <id>]
 *            [priority <priority>]
 *            [adhoc <on|off>]
 *            [outfile <name>]
 *            [file <name>] [cycles <count>] [autoreload <on|off>] 
 */
static int command_source_add(client_t *client, char *buf)
{
  scamper_source_params_t ssp;
  char *params[24];
  int   i, cnt = sizeof(params) / sizeof(char *);
  char *file = NULL, *name = NULL, *priority = NULL, *adhoc = NULL;
  char *descr = NULL, *list_id = NULL, *cycles = NULL, *autoreload = NULL;
  char *outfile = NULL, *command = NULL, *cycle_id = NULL;
  long  l;
  char *next;
  param_t handlers[] = {
    {"adhoc",      &adhoc},    
    {"autoreload", &autoreload},
    {"command",    &command},
    {"cycle_id",   &cycle_id},
    {"cycles",     &cycles},
    {"descr",      &descr},
    {"file",       &file},
    {"list_id",    &list_id},
    {"name",       &name},
    {"outfile",    &outfile},    
    {"priority",   &priority},
  };
  int handler_cnt = sizeof(handlers) / sizeof(param_t);

  if(params_get(buf, params, &cnt) == -1)
    {
      client_send(client, "ERR source add params_get failed");
      return -1;
    }

  for(i=0; i<cnt; i += 2)
    {
      if(i+1 != cnt) next = params[i+1];
      else next = NULL;

      if(param_handler(handlers, handler_cnt, client, params[i], next) == -1)
	{
	  client_send(client, "ERR source add param '%s' failed", params[i]);
	  return -1;
	}
    }

  /*
   * initialise the scamper_source_params struct with suitable default
   * values that will be passed to scamper_addresslist_sourceadd
   */
  ssp.list_id    = 0;
  ssp.cycle_id   = 1;
  ssp.name       = name;
  ssp.descr      = descr;
  ssp.command    = command;
  ssp.priority   = 1;
  ssp.adhoc      = 0;
  ssp.sof        = NULL;
  ssp.filename   = file;
  ssp.cycles     = 1;
  ssp.autoreload = 0;

  if(name == NULL)
    {
      client_send(client, "ERR required parameter 'name' missing");
      return -1;
    }

  /* sanity check the adhoc parameter */
  if(adhoc != NULL && get_switch(client, "adhoc", adhoc, &l) == -1)
    {
      return -1;
    }
  ssp.adhoc = l;

  /* sanity check the autoreload parameter */
  if(autoreload != NULL &&
     get_switch(client, "autoreload", autoreload, &l) == -1)
    {
      return -1;
    }
  ssp.autoreload = l;

  /* sanity check the cycle parameter */
  if(cycles != NULL)
    {
      if(string_tolong(cycles, &l) == -1 || l < 0)
	{
	  client_send(client, "ERR cycle <number gte 0>");
	  return -1;
	}
      ssp.cycles = l;
    }

  /* sanity check the priority parameter */
  if(priority != NULL)
    {
      if(string_tolong(priority, &l) == -1 || l < 0 || l > 0x7fffffff)
	{
	  client_send(client, "ERR priority <number gte 0>");
	  return -1;
	}
      ssp.priority = l;
    }

  /* sanity check the list_id parameter */
  if(list_id != NULL)
    {
      if(string_tolong(list_id, &l) == -1 || l < 0 || l > 0x7fffffffL)
	{
	  client_send(client, "ERR list_id <number gte 0>");
	  return -1;
	}
      ssp.list_id = l;
    }

  /* sanity check the cycle_id parameter */
  if(cycle_id != NULL)
    {
      if(string_tolong(cycle_id, &l) == -1 || l < 0 || l > 0x7fffffffL)
	{
	  client_send(client, "ERR cycle_id <number gte 0>");
	  return -1;
	}
      ssp.cycle_id = l;
    }

  /* look up the outfile's name */
  if(outfile != NULL)
    {
      if((ssp.sof = scamper_outfiles_get(outfile)) == NULL)
	{
	  client_send(client, "ERR unknown outfile '%s'");
	  return -1;
	}
    }

  if(scamper_addresslist_addsource(&ssp) == NULL)
    {
      client_send(client, "ERR could not add source");
      return -1;
    }

  client_send(client, "OK source added");

  return 0;
}

/*
 * command_source_cycle
 *
 * source cycle <name>
 */
static int command_source_cycle(client_t *client, char *buf)
{
  scamper_source_t *source;
  char *params[1];
  char *name;
  int   cnt = sizeof(params) / sizeof(char *);

  if(params_get(buf, params, &cnt) == -1)
    {
      client_send(client, "ERR source cycle params_get failed");
      return -1;
    }

  if(cnt != 1)
    {
      client_send(client, "ERR missing required parameter for source cycle");
      return -1;
    }

  name = params[0];
  if((source = scamper_addresslist_getsource(name)) == NULL)
    {
      client_send(client, "ERR no source '%s'", name);
      return -1;
    }

  if(scamper_addresslist_cyclesource(source) == -1)
    {
      client_send(client, "ERR could not cycle source '%s'", name);
      return -1;
    }

  client_send(client, "OK");

  return 0;
}

/*
 * command_source_delete
 *
 * source delete <name>
 */
static int command_source_delete(client_t *client, char *buf)
{
  scamper_source_t *source;
  char *name;
  char *params[1];
  int   cnt = sizeof(params) / sizeof(char *);

  if(params_get(buf, params, &cnt) == -1)
    {
      client_send(client, "ERR source delete params_get failed");
      return -1;
    }

  if(cnt != 1)
    {
      client_send(client, "ERR missing required parameter for source delete");
      return -1;
    }

  name = params[0];

  if((source = scamper_addresslist_getsource(name)) == NULL)
    {
      client_send(client, "ERR unknown source '%s'", params[0]);
      return -1;
    }

  if(scamper_addresslist_delsource(source) == -1)
    {
      client_send(client, "ERR could not delete source '%s'", name);
      return -1;
    }

  client_send(client, "OK source '%s' deleted", name);

  return 0;
}

static int source_foreach(void *param, scamper_source_t *source)
{
  client_t *client = (client_t *)param;
  char str[1024];

  client_send(client, "INFO %s", source_tostr(str, sizeof(str), source));

  return 0;
}

/*
 * command_source_list
 *
 * source list [<name>]
 *
 */
static int command_source_list(client_t *client, char *buf)
{
  scamper_source_t *source;
  char *params[1], str[1024];
  char *name;
  int   cnt = sizeof(params) / sizeof(char *);

  /* if there is no parameter, then dump all lists */
  if(buf == NULL)
    {
      scamper_addresslist_foreach(client, source_foreach);
      client_send(client, "OK");
      return 0;
    }

  /* if there is a parameter, then use that to find a source */
  if(params_get(buf, params, &cnt) == -1 || cnt != 1)
    {
      client_send(client, "ERR source check params_get failed");
      return -1;
    }
  name = params[0];
  if((source = scamper_addresslist_getsource(name)) == NULL)
    {
      client_send(client, "ERR no source '%s'", name);
      return 0;
    }
  client_send(client, "INFO %s", source_tostr(str, sizeof(str), source));
  client_send(client, "OK");

  return 0;
}

/*
 * command_source_update
 *
 * source update <name> [autoreload <on|off>] [cycles <count>]
 *                      [priority <priority>]
 *
 */
static int command_source_update(client_t *client, char *buf)
{
  scamper_source_t *source;
  char *autoreload = NULL, *cycles = NULL, *priority = NULL;
  long  l_autoreload, l_cycles, l_priority;
  int   i_autoreload, i_cycles, i_priority;
  param_t handlers[] = {
    {"autoreload", &autoreload},
    {"cycles",     &cycles},
    {"priority",   &priority},
  };
  int handler_cnt = sizeof(handlers) / sizeof(param_t);  
  char *params[10];
  int   i, cnt = sizeof(params) / sizeof(char *);
  char *next;

  if(buf == NULL)
    {
      client_send(client, "ERR missing name parameter");
      return 0;
    }

  if(params_get(buf, params, &cnt) == -1)
    {
      client_send(client, "ERR source update params_get failed");
      return -1;
    }

  /* the name parameter should be in parameter zero */
  if(cnt < 1)
    {
      client_send(client, "ERR missing name parameter");
      return 0;
    }

  /* find the source */
  if((source = scamper_addresslist_getsource(params[0])) == NULL)
    {
      client_send(client, "ERR no such source '%s'", params[0]);
      return 0;
    }

  /* parse out each parameter */
  for(i=1; i<cnt; i += 2)
    {
      if(i+1 != cnt) next = params[i+1];
      else next = NULL;

      if(param_handler(handlers, handler_cnt, client, params[i], next) == -1)
	{
	  client_send(client, "ERR source update param '%s' failed",params[i]);
	  return -1;
	}
    }

  /*
   * sanity check that the autoreload or cycle parameters make sense
   * for this type of source
   */
  if(scamper_source_getfilename(source) == NULL &&
     (autoreload != NULL || cycles != NULL))
    {
      client_send(client,"ERR can't specify autoreload/cycles on adhoc source");
      return 0;
    }

  /* sanity check / parse the parameters to autoreload, cycle, priority */
  if(autoreload != NULL &&
     get_switch(client, "autoreload", autoreload, &l_autoreload) == -1)
    {
      client_send(client, "ERR autoreload <on|off>");
      return 0;
    }
  if(cycles != NULL &&
     (string_tolong(cycles, &l_cycles) == -1 || l_cycles < 0))
    {
      client_send(client, "ERR cycles <number gte 0>");
      return 0;
    }
  if(priority != NULL &&
     (string_tolong(priority, &l_priority) == -1 || l_priority < 0))
    {
      client_send(client, "ERR priority <number gte 0>");
      return 0;
    }

  i_autoreload = l_autoreload;
  i_cycles     = l_cycles;
  i_priority   = l_priority;

  /* now that the parameters have been sanity checked, use them */
  if(scamper_source_update(source,
			   (autoreload != NULL ? &i_autoreload : NULL),
			   (cycles     != NULL ? &i_cycles     : NULL),
			   (priority   != NULL ? &i_priority   : NULL)))
  {
    client_send(client, "ERR could not update source %s",
		scamper_source_getname(source));
    return 0;
  }

  client_send(client, "OK");
  return 0;
}

static int command_source(client_t *client, char *buf)
{
  static command_t handlers[] = {
    {"add",    command_source_add},
    {"cycle",  command_source_cycle},
    {"delete", command_source_delete},
    {"list",   command_source_list},
    {"update", command_source_update},
  };
  static int handler_cnt = sizeof(handlers) / sizeof(command_t);
  char *next;
  int ret;

  if(buf == NULL)
    {
      client_send(client,
		  "ERR usage: source [add | cycle | delete | list | update]");
      return 0;
    }

  next = string_nextword(buf);
  if(command_handler(handlers, handler_cnt, client, buf, next, &ret) == -1)
    {
      client_send(client, "ERR unhandled command '%s'", buf);
      return 0;
    }

  return 0;
}

static int command_shutdown_cancel(client_t *client, char *buf)
{
  scamper_exitwhendone(0);
  client_send(client, "OK");
  return 0;
}

static int command_shutdown_done(client_t *client, char *buf)
{
  scamper_exitwhendone(1);
  client_send(client, "OK");
  return 0;
}

static int command_shutdown_flush(client_t *client, char *buf)
{
  /* empty the address list of all sources */
  scamper_addresslist_empty();

  /* tell scamper to exit when it has finished probing the existing window */
  scamper_exitwhendone(1);

  client_send(client, "OK");
  return 0;
}

static int command_shutdown_now(client_t *client, char *buf)
{
  /* empty the active trace window */
  scamper_queue_empty();

  /* empty the address list of all sources */
  scamper_addresslist_empty();

  /* tell scamper to exit when it has finished probing the existing window */
  scamper_exitwhendone(1);

  client_send(client, "OK");

  return 0;
}

static int command_shutdown(client_t *client, char *buf)
{
  static command_t handlers[] = {
    {"cancel", command_shutdown_cancel},
    {"done",   command_shutdown_done},
    {"flush",  command_shutdown_flush},
    {"now",    command_shutdown_now},
  };
  static int handler_cnt = sizeof(handlers) / sizeof(command_t);
  char *next;
  int ret;

  if(buf == NULL)
    {
      client_send(client, "ERR usage: [cancel | done | flush | now]");
      return 0;
    }

  next = string_nextword(buf);
  if(command_handler(handlers, handler_cnt, client, buf, next, &ret) == -1)
    {
      client_send(client, "ERR unhandled command '%s'", buf);
      return 0;
    }

  return 0;
}

static int client_read_line(void *param, uint8_t *buf, size_t len)
{
  static command_t handlers[] = {
    {"do",         command_do},
    {"exit",       command_exit},
    {"get",        command_get},
    {"help",       command_help},
    {"observe",    command_observe},
    {"outfile",    command_outfile},
    {"set",        command_set},
    {"shutdown",   command_shutdown},
    {"source",     command_source},
  };
  static int handler_cnt = sizeof(handlers) / sizeof(command_t);
  client_t *client = (client_t *)param;
  char *next;
  int ret;

  /* make sure all the characters in the string are printable */
  if(string_isprint((char *)buf, len) == 0)
    {
      client_send(client, "ERR invalid character in line");
      client_free(client);
      return 0;
    }

  /* XXX: should check for null? */
  next = string_nextword((char *)buf);

  if(command_handler(handlers,handler_cnt,client,(char *)buf,next,&ret) == -1)
    {
      client_send(client, "ERR unhandled command '%s'", buf);
      return 0;
    }

  return 0;
}

static void client_read(const int fd, void *param)
{
  client_t *client;
  ssize_t rc;
  uint8_t buf[256];

  client = (client_t *)param;
  assert(scamper_fd_fd_get(client->fdn) == fd);

  /* try and read more from the client */
  if((rc = read(fd, buf, sizeof(buf))) < 0)
    {
      if(errno != EAGAIN)
	{
	  printerror(errno, strerror, __func__, "read failed");
	}
      return;
    }

  /* if the client has disconnected then yank it */
  if(rc == 0)
    {
      client_free(client);
      return;
    }

  scamper_linepoll_handle(client->lp, buf, (size_t)rc);

  return;
}

/*
 * client_alloc
 *
 * given a new inbound client, allocate a new node
 *
 */
static client_t *client_alloc(struct sockaddr *sa, socklen_t slen, int fd)
{
  client_t *client;

  /* make the socket non-blocking, so a read or write will not hang scamper */
  if(fcntl_set(fd, O_NONBLOCK) == -1)
    {
      return NULL;
    }

  /* allocate the structure that holds the socket/client together */
  if((client = malloc_zero(sizeof(struct client))) == NULL)
    {
      return NULL;
    }

  /* put the node into the list of sockets that are connected */
  if((client->node = dlist_tail_push(client_list, client)) == NULL)
    {
      goto cleanup;
    }

  /* make a copy of the sockaddr that connected to scamper */
  if((client->sa = malloc(slen)) == NULL)
    {
      goto cleanup;
    }
  memcpy(client->sa, sa, slen);

  /* add the file descriptor to the event manager */
  if((client->fdn=scamper_fd_private(fd,client_read,client,NULL,NULL)) == NULL)
    {
      goto cleanup;
    }

  /* put a wrapper around the socket to read from it one line at a time */
  if((client->lp = scamper_linepoll_alloc(client_read_line, client)) == NULL)
    {
      goto cleanup;
    }

  if((client->wb = scamper_writebuf_alloc(client->fdn)) == NULL)
    {
      goto cleanup;
    }

  return client;

 cleanup:
  if(client->wb != NULL) scamper_writebuf_free(client->wb);
  if(client->lp != NULL) scamper_linepoll_free(client->lp, 0);
  if(client->node != NULL) dlist_node_pop(client_list, client->node);
  if(client->sa != NULL) free(client->sa);
  free(client);

  return NULL;
}

static void control_accept(const int fd, void *param)
{
  struct sockaddr_storage ss;
  client_t *client;
  socklen_t socklen;
  int s;

  /* accept the new client */
  socklen = sizeof(ss);
  if((s = accept(fd, (struct sockaddr *)&ss, &socklen)) == -1)
    {
      return;
    }

  scamper_debug(__func__, "fd %d", s);

  /* allocate a client struct to keep track of data coming in on socket */
  if((client = client_alloc((struct sockaddr *)&ss, socklen, s)) == NULL)
    {
      shutdown(s, SHUT_RDWR);
      close(s);
      return;
    }

  return;
}

int scamper_control_init(int port)
{
  struct sockaddr_in sin;
  struct in_addr     in;
  int                fd = -1, opt;

  /* open the TCP socket we are going to listen on */
  if((fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) == -1)
    {
      printerror(errno, strerror, __func__, "could not create TCP socket");
      return -1;
    }

  opt = 1;
  if(setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) != 0)
    {
      printerror(errno, strerror, __func__, "could not set SO_REUSEADDR");
      goto cleanup;
    }

  /* bind the socket to loopback on the specified port */
  in.s_addr = htonl(INADDR_LOOPBACK);
  sockaddr_compose((struct sockaddr *)&sin, AF_INET, &in, port);
  if(bind(fd, (struct sockaddr *)&sin, sizeof(sin)) == -1)
    {
      printerror(errno, strerror, __func__,
		 "could not bind to loopback port %d", port);
      goto cleanup;
    }

  /* tell the system we want to listen for new clients on this socket */
  if(listen(fd, -1) == -1)
    {
      printerror(errno, strerror, __func__, "could not listen");
      goto cleanup;
    }

  /* allocate the list of clients connected to this scamper process */
  if((client_list = dlist_alloc()) == NULL)
    {
      goto cleanup;
    }

  if((fdn = scamper_fd_private(fd, control_accept, NULL, NULL, NULL)) == NULL)
    {
      goto cleanup;
    }

  return 0;

 cleanup:
  if(client_list != NULL) dlist_free(client_list);
  if(fdn != NULL) scamper_fd_free(fdn);
  close(fd);
  return -1;
}

/*
 * scamper_control_cleanup
 *
 * go through and free all the clients that are connected.
 * write anything left in the writebuf to the clients (non-blocking) and
 * then close the socket.
 */
void scamper_control_cleanup()
{
  client_t *client;
  int fd;

  if(client_list != NULL)
    {
      while((client = dlist_head_pop(client_list)) != NULL)
	{
	  client->node = NULL;
	  scamper_writebuf_flush(client->wb);
	  client_free(client);
	}

      dlist_free(client_list);
      client_list = NULL;
    }

  /* stop monitoring the control socket for new connections */
  if(fdn != NULL)
    {
      if((fd = scamper_fd_fd_get(fdn)) != -1)
	{
	  close(fd);
	}

      scamper_fd_free(fdn);
      fdn = NULL;
    }

  return;
}