# include "pantasks.h"
# define DEBUG 0

static int NCLIENTS;
static int Nclients;
static int *clients;
static IOBuffer **buffers;

void InitClients () {

  Nclients = 0;
  NCLIENTS = 10;
  ALLOCATE (clients, int, NCLIENTS);
  ALLOCATE (buffers, IOBuffer *, NCLIENTS);
}

/* add this client to the client table, create an IOBuffer for it */
void AddNewClient (int client) {

  if (DEBUG) gprint (GP_LOG, "adding a new client (%d)\n", client);
  clients[Nclients] = client;
  ALLOCATE (buffers[Nclients], IOBuffer, 1);
  InitIOBuffer(buffers[Nclients], 256);
  Nclients ++;
  if (Nclients >= NCLIENTS - 1) {
    NCLIENTS += 10;
    REALLOCATE (clients, int, NCLIENTS);
    REALLOCATE (buffers, IOBuffer *, NCLIENTS);
  }
}

/* add this client to the client table, create an IOBuffer for it */
int DeleteClient (int client) {

  int i, j;

  if (DEBUG) gprint (GP_LOG, "deleting a client (%d)\n", client);
  for (i = 0; i < Nclients; i++) {
    if (clients[i] == client) {
      FreeIOBuffer (buffers[i]);
      free (buffers[i]);
      close (clients[i]);
      for (j = i; j < Nclients - 1; j++) {
	clients[j] = clients[j+1];
	buffers[j] = buffers[j+1];
      }
      Nclients --;
      if ((Nclients > 10) && (Nclients / 2 < NCLIENTS)) {
	NCLIENTS = Nclients + 10;
	REALLOCATE (clients, int, NCLIENTS);
	REALLOCATE (buffers, IOBuffer *, NCLIENTS);
      }
      return TRUE;
    }
  }
  // did not find the client
  return FALSE;
}

/* select for messages from the current clients; wait for 0.5s before updating client list */
void *ListenClients (void *data) {
  
  int i, Ncurrent, Nmax, status, Nread;
  char *line, log_stdout[128], log_stderr[128];
  fd_set fdSet;
  struct timeval timeout;
  IOBuffer *outbuffer;

  InitClients ();
  gprintInit ();  // each thread needs to init the printing system

  // define server output log files
  if (VarConfig ("PANTASKS_SERVER_STDOUT", "%s", log_stdout) != NULL) {
    gprintSetFileThisThread (GP_LOG, log_stdout);
  } else {
    strcpy (log_stdout, "stdout");
  }
  if (VarConfig ("PANTASKS_SERVER_STDERR", "%s", log_stderr) != NULL) {
    gprintSetFileThisThread (GP_ERR, log_stderr);
  } else {
    strcpy (log_stderr, "stderr");
  }

  while (1) {

    /* Wait up to 0.5 second - need to timeout to update client list */
    /* timeout gets mucked: need to reset before each select */
    timeout.tv_sec = 0;
    timeout.tv_usec = 500000;

    /* place all of the clients in the fdSet */
    Ncurrent = Nclients;
    Nmax = 0;
    FD_ZERO (&fdSet);
    for (i = 0; i < Ncurrent; i++) {
      Nmax = MAX (Nmax, clients[i]);
      FD_SET (clients[i], &fdSet);
    }    
    Nmax ++;

    /* block until we have some data on the pipes (or timeout) */
    if (DEBUG) gprint (GP_ERR, "listening to %d clients\n", Ncurrent);
    status = select (Nmax, &fdSet, NULL, NULL, &timeout);
    if (status == -1) {
      perror("select()");
      return (FALSE);
    }

    /* if no data, update client list, wait for another select */
    if (status <= 0) continue;

    /* loop over the clients with data */
    for (i = 0; i < Ncurrent; i++) {
      /* if client has no data, skip it */
      if (!FD_ISSET(clients[i], &fdSet)) continue;

      /* read until the pipe is empty: 0 is closed, -1 is empty, -2 is error */
      Nread = 1;
      while (Nread > 0) {
	Nread = ReadtoIOBuffer (buffers[i], clients[i]);	
      }
      if ((Nread == 0) || (Nread == -2)) {
	/* error: do something */
	if (DEBUG && (Nread == 0)) gprint (GP_ERR, "socket is closed\n");
	if (DEBUG && (Nread == -2)) gprint (GP_ERR, "error reading from socket\n");
	DeleteClient (clients[i]);
	continue;
      }

      if (DEBUG) gprint (GP_ERR, "read %d total bytes\n", buffers[i][0].Nbuffer);

      /* see if we have a complete message waiting; if not, keep waiting for messages */
      line = CheckForMessage (buffers[i]);
      if (line == NULL) continue;

      /* we now have a possible command from the client: run it */
      /* in this thread, we set the print output destination to be an
	 internal buffer, which we dump at the end of the execution */
      /* the commands sent to the server should not have ; */
      stripwhite (line);
      if (*line) {

	/* set buffers for the output for this client */
	gprintSetBuffer (GP_LOG);
	gprintSetBuffer (GP_ERR);

	/* run the command, return the exit status */
	status = multicommand (line);
	SendMessage (clients[i], "STATUS %d", status);

	// return the stderr messages first
	outbuffer = gprintGetBuffer (GP_ERR);
	if (outbuffer) {
	  SendMessageFixed (clients[i], outbuffer[0].Nbuffer, outbuffer[0].buffer);
	} else {
	  SendMessageFixed (clients[i], 0, "");
	}	  

	// return the stdout messages first
	outbuffer = gprintGetBuffer (GP_LOG);
	if (outbuffer) {
	  SendMessageFixed (clients[i], outbuffer[0].Nbuffer, outbuffer[0].buffer);
	} else {
	  SendMessageFixed (clients[i], 0, "");
	}	  
	
	/* clear and reset the output buffers to their last output file names */
	gprintSetFileAllThreads (GP_LOG, NULL);
	gprintSetFileAllThreads (GP_ERR, NULL);
      }
      free (line);
    }
    /* if Nread == -2, we probably need to drop the client */
    /* check if we need to drop / remove any clients */
    /* check if we need to shut down the thread */
  }
}

/* the AddClient commands are issued by the parent thread
   the value of Nclients may increase after we check it here. 
   only this thread is allowed to decrease Nclients and remove
   a client from the table */
