Copy data from agent to viewer

	The [[start_request]] function is called by [[W3Aprocess]]
	and [[W3Asubprocess]] to read data from an agent, open a
	viewer and copy the data to that viewer. Because we don't want
	to block while the agent is reading data, the reading is done
	in input-handler callbacks.

	There are three of those: [[get_peek_cb]] is called until the
	agent reports that it has data to read; [[get_info_cb]] is
	called until the agent has returned the MIME type of the data;
	[[get_input_cb]] is called repeatedly until all the data has
	been read.

	Requests are queued, so that only a small number of them
	execute concurrently. An input handler that stops
	automatically starts the next request from the queue.

	For extra-high priority requests (usually requests that are
	the result of the user clicking a hyperlink), extra
	connections are made above the [[MAX_ACTIVE]] that are
	normally allowed concurrently. A maximum of [[MAX_RESERVE]]
	additional connections is opened for them.

<<*>>=
#include "config.h"
#include <fcntl.h>
#include <Xm/Xm.h>
#include "w3a.h"
#include "str.h"
#include "url.h"
#include "rcfile.e"
#include "globals.e"
#include "dynload.e"
#include "requests.e"
#include "gui.e"
#include "cache.e"
#include "proxy.e"

#define ALARM 0
#if ALARM
#include <signal.h>
#endif


static int nr_active = 0;			/* Concurrent requests */
static Bool progress_in_use[MAX_ACTIVE];
static Bool progress_inited = FALSE;


typedef struct _Queue {
    struct _Queue *next;
    ViewerInfo *info;
    URI uri;					/* Parsed URL */
} *Queue;

static Queue last, first = NULL;


#define queue_empty() (first == NULL)


/* enqueue -- add a request to the queue, possibly at the head of the queue */
static void enqueue(URI uri, ViewerInfo *info, Bool in_front)
{
    Queue q;

    new(q);
    q->info = info;
    q->uri = uri;
    if (in_front) {				/* Prioritized request */
	q->next = first;
	if (! first) last = q;
	first = q;
    } else {					/* Normal request */
	q->next = NULL;
	if (! first) first = q; else last->next = q;
	last = q;
    }
}


/* dequeue -- get a request from the queue (caller must free url)*/
static ViewerInfo *dequeue(URI *uri)
{
    Queue q;
    ViewerInfo *v;

    assert(first != NULL && last != NULL);
    q = first;
    v = q->info;
    *uri = q->uri;
    first = first->next;
    dispose(q);
    return v;
}


#if ALARM
/* alarm_handler -- just reinstall the alarm */
static void alarm_handler(int sig)
{
    (void) signal(SIGALRM, alarm_handler);
    (void) alarm(2);
    /* Doesn't do anything now, maybe could update screen? */
}
#endif


/* Show error message and remove viewer from screen */
static void error_and_remove_doc(const char *s, ViewerInfo *info)
{
    int n;

    show_error(s, *info->doc);
    if ((n = find_among_tabs(info->doc->url)) >= 0) remove_doc(n);
    else (void) free_request_rec(info);
}


static void write_cb(XtPointer client_data, int *fd, XtInputId *id);
/* forward */

static void get_info_cb(XtPointer client_data, int *fd, XtInputId *id);
/* forward */

/* try_execute_next_request -- start next request, if any and not too many */
static void try_execute_next_request(void)
{
#ifdef DEBUG
    static int total_requests = 0, from_cache = 0;
#endif
    ViewerInfo *info;
    const char *where;
    size_t size;
    URI uri;
    int i;

    /* First try to satisfy requests from the cache */
    for (;;) {
	if (queue_empty()) return;		/* No requests pending */
	info = dequeue(&uri);			/* Get next request */
	if (info->method != GET_METHOD) break;	/* Cannot be cached */
	if (! in_cache(info->doc, &where, &size)) break;

	/* The request can be satisfied from the cache */
#if 0
#ifdef DEBUG
	total_requests++;
	from_cache++;
	debug("cache hits = %d/%d - - url=%s\n",
	      from_cache, total_requests, info->doc->url);
#endif
#endif
	info->viewer = find_viewer(info->doc->mime_type);
	assert(info->viewer != -1);		/* Can't be cached otherwise */
	if (! viewer_open(info->viewer, *info->doc, info->w, (long) info)
	    || viewer_write(info->viewer, (long) info, where, size) != size
	    || viewer_write(info->viewer, (long) info, NULL, 0) != 0) {
	    error_and_remove_doc("[try_execute_next_request]", info);
	    return;
	}
	if (info->toplevel) set_tab_titles();
    }

    /* Now we have a request `info' that isn't cached, check # connections */
    if (nr_active == MAX_ACTIVE + MAX_RESERVE
	|| (nr_active >= MAX_ACTIVE && ! info->high_prio)) {
	enqueue(uri, info, TRUE);		/* Put request back */
	return;					/* Give up */
    }

    /* There is room for a new connection */
    if (info->fd < 0) {				/* PUT/POST already opened */
	char *newurl = NULL, *url = info->doc->url;
	Bool pp;
	assert(info->method == GET_METHOD);
	pp = insert_proxy(uri, url, &newurl);
	info->agent = find_agent(pp ? "proxy" : strip2str(uri.scheme));
	info->fd = agent_open(info->agent, pp ? newurl : url, GET_METHOD,
			      O_NONBLOCK, info->doc->referer);
	if (pp) dispose(newurl);
    }
    if (info->fd < 0) {
	error_and_remove_doc("[try_execute_next_request]", info);
	return;
    }
    if (info->toplevel) {			/* Temp file for Save cmd */
	(void) tmpnam(info->tmpnam);
	info->tmpfd = open(info->tmpnam, O_WRONLY|O_CREAT, 0600);
    }
    info->read_id = XtAppAddInput
	(app_context, info->fd, (XtPointer)XtInputReadMask, get_info_cb, info);
    info->write_id = XtAppAddInput
	(app_context, info->fd, (XtPointer) XtInputWriteMask, write_cb, info);
#if 1
    if (info->toplevel) {			/* Only show bar for docs */
	info->show_progress = 0;
	show_progress(info, -1);		/* Clear percentage bar */
    }
#else
    for (i = 0; i < MAX_ACTIVE; i++)		/* Find free percentage bar */
	if (! progress_in_use[i]) {
	    info->show_progress = i;
	    progress_in_use[i] = TRUE;
	    show_progress(info, -1);		/* Clear percentage bar */
	    break;
	}
#endif
    nr_active++;
    XtSetSensitive(interrupt, TRUE);		/* `Interrupt' button */
#if 0
    XFlush(XtDisplay(toplevel));
#endif
#if 0
#ifdef DEBUG
    total_requests++;
    debug("cache hits=%d/%d #active=%d url=%s\n",
	  from_cache, total_requests, nr_active, info->doc->url);
#endif
#endif
#if ALARM
    (void) alarm(2);
#endif
}
@

	|stop_request| is called when an input handler detects that
	the user has pressed the interrupt button, or when the
	requests has run to its end. It removes the input handler and
	tries to execute the next request.

<<*>>=
/* stop_request -- remove input handler, start the next */
static void stop_request(ViewerInfo *v)
{
    alarm(0);
    assert(v);
#if 1
    if (v->show_progress >= 0) show_progress(v, -1);
#endif
    if (v->write_id) {XtRemoveInput(v->write_id); v->write_id = NULL;}
    if (v->read_id) {XtRemoveInput(v->read_id); v->read_id = NULL;}
    assert(v->agent == -1 || v->agent >= 0);
    if (v->agent >= 0) {agent_close(v->agent, v->fd); v->agent = -1;}
    if (v->tmpfd >= 0) {(void) close(v->tmpfd); v->tmpfd = -1;}
    if (v->show_progress >= 0) progress_in_use[v->show_progress] = FALSE;
    if (v->toplevel && XtIsRealized(v->w))
	XDefineCursor(XtDisplay(v->w), XtWindow(v->w), None);
    nr_active--;
    try_execute_next_request();
}


/* show_get_error -- show error message and remove input handler */
static void show_get_error(int err, const char *s, ViewerInfo *info)
{
    stop_request(info);
    errno = err;
    error_and_remove_doc(s, info);
}
@

	The [[get_input_cb]] function is called by Xt whenever there
	is anything to read on socket [[*fd]]. The task of the
	function is to copy incoming data from the agent to the viewer.

<<*>>=
/* get_input_cb -- callback attached to an agent's file descriptor */
static void get_input_cb(XtPointer client_data, int *fd, XtInputId *id)
{
    ViewerInfo *v = (ViewerInfo *) client_data;
    char buf[BUFSIZ];
    int n;

    assert(v);
    assert(v->read_id == *id);
    assert(v->fd == *fd);
    if (v->interrupted) {			/* Disabled for some reason */
	stop_request(v);
	return;
    }
    if ((n = agent_read(v->agent, *fd, buf, sizeof(buf))) == -1) {
	if (errno != EAGAIN) show_get_error(errno, "[get_input_cb]", v);
    } else {					/* Valid input */
	if (v->show_progress >= 0) show_progress(v, n);
	if (v->cache_id >= 0) store_in_cache(v->cache_id, buf, n);
	if (v->tmpfd >= 0) (void) write(v->tmpfd, buf, n);

	if (v->viewer != -1) {
	    /* If v->viewer == -1 we read the data and ignore it */
	    n = viewer_write(v->viewer, (long) v, buf, n);
	    if (n == -1) show_get_error(errno, "[get_input_cb]", v);
	    if (n == 0) {			/* End of data */
		/* Get title again, this time from viewer */
		viewer_info(v->viewer, (long) v, v->doc);
		if (v->toplevel) set_tab_titles();
		/* Notify all applets of the new document again */
		if (v->need_event) W3Aevent(0, NEW_DOCUMENT, v->doc);
	    }
	}
	if (n == 0) {
	    stop_request(v);
	    if (v->show_progress >= 0) show_progress_ready(v);
	}
    }
}
@

<<*>>=
/* write_cb -- callback attached to an agent's file descriptor */
static void write_cb(XtPointer client_data, int *fd, XtInputId *id)
{
    ViewerInfo *v = (ViewerInfo *) client_data;

    assert(v);
    assert(v->write_id == *id);
    assert(v->fd == *fd);
    if (v->interrupted) {			/* Interupted by user */
	stop_request(v);
    } else if (agent_done(v->agent, *fd)) {	/* Agent is ready sending */
	XtRemoveInput(v->write_id);
	v->write_id = NULL;
    } else if (errno != EAGAIN) {		/* An error occured */
	show_get_error(errno, "[write_cb]", v);
    }
}
@

	The [[get_info_cb]] function is called when there is data
	available, but the type of the data is not yet know. It asks
	the agent for the MIME type of the data. If that information
	is not yet available, the function does nothing. The next time
	it is called it tries again.

	If the agent returns a status code starting with `3', it means
	the document can be found under a different URL. If the full
	status code is `301' the old URL should be discarded,
	otherwise it is a temporary redirect. In the latter case the
	[[ViewerInfo]] is left unchanged and when the request is
	started again, it will use [[location]] rather than [[url]].

	In the case of `301', the old URL is removed and the
	`location' is put in its place before the request is restarted.

<<*>>=
/* get_info_cb -- callback attached to an agent's file descriptor */
static void get_info_cb(XtPointer client_data, int *fd, XtInputId *id)
{
    ViewerInfo *v = (ViewerInfo *) client_data;
    char buf[BUFSIZ];
    URI uri;
    int n;

    assert(v);
    assert(v->read_id == *id);
    assert(v->fd == *fd);
    if (v->interrupted) {			/* Interupted by user */
	stop_request(v);
	return;
    }
    if (! agent_info(v->agent, *fd, v->doc)) {
	if (errno != EAGAIN) show_get_error(errno, "[get_info_cb]", v);
    } else {					/* Got info */
	XtRemoveInput(v->read_id); v->read_id = NULL;

	if (v->doc->status && v->doc->status[0] == '3' && v->doc->location) {

	    /* 3xx = URL redirection */
#if 0
	    debug("Found redirect from %s to %s\n",
		  v->doc->url, v->doc->location);
#endif
	    stop_request(v);
	    dispose(v->doc->mime_type);		/* Reset info fields */
	    dispose(v->doc->mime_params);
	    dispose(v->doc->title);
	    v->doc->size = -1;
	    v->fd = -1;
	    if (! URL_parse(v->doc->location, &uri) || uri.tp != URI_URL) {
		show_get_error(EURL, "[get_info_cb]", v); /* Error new URL */
		return;
	    }
	    dispose(v->doc->url);		/* Move location to url */
	    dispose(v->doc->location);
	    v->doc->url = uri2str(uri);
	    dispose(v->doc->status);
	    v->agent= find_agent(strip2str(uri.scheme));
	    enqueue(uri, v, TRUE);		/* Prioritized */
	    try_execute_next_request();

	} else if (v->doc->status && n_eq(v->doc->status, "401", 3)) {

	    /* Password required */
	    show_get_error(ENYI, "(Document required password/authentication\n\
which is not implemented yet)", v);

	} else if (v->doc->status && v->doc->status[0] != '2') {

	    /* Server reported an error */
	    /* show_get_error("[get_info_cb]", v); */
	    show_get_error(EURL, v->doc->status, v);
	    return;

	} else if (eq(v->doc->mime_type, "none")) {

	    /* Agent returns success, but no document */
	    stop_request(v);
	    if ((n = find_among_tabs(v->doc->url)) >= 0)
		remove_doc(n);
	    
	} else {
	
	    /* Now we know the MIME type, open the appropriate viewer */
	    if ((v->viewer = find_viewer(v->doc->mime_type)) == -1) {
		show_get_error(ETYPE, "[get_info_cb]", v);
		/* TO DO: offer to save to file */
		return;
	    }
	    if (! viewer_open(v->viewer, *v->doc, v->w, (long) v)) {
		show_get_error(errno, "[get_info_cb]", v);
		v->viewer = -1;
		return;
	    }

	    /* Notify all applets of the new document */
	    if (v->need_event) W3Aevent(0, NEW_DOCUMENT, v->doc);

	    /* Install get_input_cb as new input handler */
	    v->cache_id = new_cache_id(*v->doc);
	    v->read_id = XtAppAddInput
		(app_context, v->fd, (XtPointer) XtInputReadMask,
		 get_input_cb, v);

	    /* Remove the watch cursor (maybe too early?) */
	    if (v->toplevel && XtIsRealized(v->w))
		XDefineCursor(XtDisplay(v->w), XtWindow(v->w), None);
	}
    }
}


/* start_request -- start the request as an X input handler */
EXPORT void start_request(URI uri, ViewerInfo *info, Bool high_prio)
{
    if (! progress_inited) {
	int i;
	for (i = 0; i < MAX_ACTIVE; i++) progress_in_use[i] = FALSE;
	progress_inited = TRUE;
#if ALARM
	(void) signal(SIGALRM, alarm_handler);
#endif
    }
    enqueue(uri, info, high_prio);
    try_execute_next_request();
}