The R Project SVN R

Rev

Rev 62726 | Rev 68392 | Go to most recent revision | Details | Compare with Previous | Last modification | View Log | RSS feed

Rev Author Line No. Line
48990 urbaneks 1
/*
2
 *  R : A Computer Language for Statistical Data Analysis
59184 ripley 3
 *  Copyright (C) 2009-12 The R Core Team.
48990 urbaneks 4
 *
5
 *  This program is free software; you can redistribute it and/or modify
6
 *  it under the terms of the GNU General Public License as published by
7
 *  the Free Software Foundation; either version 2 of the License, or
8
 *  (at your option) any later version.
9
 *
10
 *  This program is distributed in the hope that it will be useful,
11
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
12
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13
 *  GNU General Public License for more details.
14
 *
15
 *  You should have received a copy of the GNU General Public License
16
 *  along with this program; if not, a copy is available at
17
 *  http://www.r-project.org/Licenses/
18
 */
19
 
20
/* This is a small HTTP server that serves requests by evaluating
21
 * the httpd() function and passing the result to the browser. */
22
 
23
/* Example:
49434 ripley 24
   httpd <- function(path,query=NULL,...) {
25
      cat("Request for:", path,"\n"); print(query);
26
      list(paste("Hello, <b>world</b>!<p>You asked for \"",path,"\".",sep=''))
27
   }
48990 urbaneks 28
   .Internal(startHTTPD("127.0.0.1",8080))
29
 */
30
 
31
/* size of the line buffer for each worker (request and header only)
32
 * requests that have longer headers will be rejected with 413 */
33
#define LINE_BUF_SIZE 1024
34
 
35
/* maximum number of active workers (parallel connections)
36
 * when exceeded the server closes new connections */
37
#define MAX_WORKERS 32
38
 
39
 
40
/* --- Rhttpd implementation --- */
41
 
42
#ifdef HAVE_CONFIG_H
43
#include <config.h>
44
#endif
45
 
46
#include <Defn.h>
47
#include <Fileio.h>
48
#include <Rconnections.h>
49
 
50
#include <stdlib.h>
51
#include <stdio.h>
52
#include <string.h>
53
 
54
#include <Rmodules/Rinternet.h>
55
 
56
#define HttpdServerActivity 8
57
#define HttpdWorkerActivity 9
58
 
59
/* this is orignally from sisock.h - system independent sockets */
60
 
61
#ifndef WIN32
49434 ripley 62
# include <R_ext/eventloop.h>
63
# include <sys/types.h>
64
# ifdef HAVE_UNISTD_H
65
#  include <unistd.h>
66
# endif
67
# ifdef HAVE_BSD_NETWORKING
68
#  include <netdb.h>
69
#  include <sys/socket.h>
70
#  include <netinet/in.h>
71
#  include <arpa/inet.h>
72
# endif
73
# include <errno.h>
48990 urbaneks 74
 
49434 ripley 75
# define sockerrno errno
76
# define SOCKET int
77
# define INVALID_SOCKET (-1)
78
# define closesocket(A) close(A)
79
# define initsocks()
80
# define donesocks()
48990 urbaneks 81
#else
49434 ripley 82
/* --- Windows-only --- */
83
# include <windows.h>
84
# include <winsock.h>
85
# include <string.h>
48990 urbaneks 86
 
49434 ripley 87
# define sockerrno WSAGetLastError()
48990 urbaneks 88
 
49434 ripley 89
# define ECONNREFUSED WSAECONNREFUSED
90
# define EADDRINUSE WSAEADDRINUSE
91
# define ENOTSOCK WSAENOTSOCK
92
# define EISCONN WSAEISCONN
93
# define ETIMEDOUT WSAETIMEDOUT
94
# define ENETUNREACH WSAENETUNREACH
95
# define EINPROGRESS WSAEINPROGRESS
96
# define EALREADY WSAEALREADY
97
# define EAFNOSUPPORT WSAEAFNOSUPPORT
98
# define EOPNOTSUPP WSAEOPNOTSUPP
99
# define EWOULDBLOCK WSAEWOULDBLOCK
100
/* those are occasionally defined by MinGW's errno, so override them
101
 * with socket equivalents */
102
# ifdef EBADF
103
#  undef EBADF
104
# endif
105
# ifdef EINVAL
106
#  undef EINVAL
107
# endif
108
# ifdef EFAULT
109
#  undef EFAULT
110
# endif
111
# ifdef EACCES
112
#  undef EACCES
113
# endif
114
# define EFAULT WSAEFAULT
115
# define EINVAL WSAEINVAL
116
# define EACCES WSAEACCES
117
# define EBADF WSAEBADF
48990 urbaneks 118
 
119
static int initsocks(void)
120
{
121
    WSADATA dt;
122
    /* initialize WinSock 1.1 */
123
    return (WSAStartup(0x0101, &dt)) ? -1 : 0;
124
}
125
 
49434 ripley 126
# define donesocks() WSACleanup()
48990 urbaneks 127
typedef int socklen_t;
128
 
49434 ripley 129
#endif /* WIN32 */
48990 urbaneks 130
 
131
/* --- system-independent part --- */
132
 
133
#define SA struct sockaddr
134
#define SAIN struct sockaddr_in
135
 
136
static struct sockaddr *build_sin(struct sockaddr_in *sa, const char *ip, int port) {
137
    memset(sa, 0, sizeof(struct sockaddr_in));
138
    sa->sin_family = AF_INET;
139
    sa->sin_port = htons(port);
140
    sa->sin_addr.s_addr = (ip) ? inet_addr(ip) : htonl(INADDR_ANY);
141
    return (struct sockaddr*)sa;
142
}
143
 
144
/* --- END of sisock.h --- */
145
 
146
/* debug output - change the DBG(X) X to enable debugging output */
147
#define DBG(X)
148
 
149
/* --- httpd --- */
150
 
151
#define PART_REQUEST 0
152
#define PART_HEADER  1
153
#define PART_BODY    2
154
 
62724 urbaneks 155
#define METHOD_POST  1
156
#define METHOD_GET   2
157
#define METHOD_HEAD  3
158
#define METHOD_OTHER 8 /* for custom requests only */
48990 urbaneks 159
 
160
/* attributes of a connection/worker */
51562 urbaneks 161
#define CONNECTION_CLOSE  0x01 /* Connection: close response behavior is requested */
162
#define HOST_HEADER       0x02 /* headers contained Host: header (required for HTTP/1.1) */
163
#define HTTP_1_0          0x04 /* the client requested HTTP/1.0 */
164
#define CONTENT_LENGTH    0x08 /* Content-length: was specified in the headers */
165
#define THREAD_OWNED      0x10 /* the worker is owned by a thread and cannot removed */
166
#define THREAD_DISPOSE    0x20 /* the thread should dispose of the worker */
167
#define CONTENT_TYPE      0x40 /* message has a specific content type set */
168
#define CONTENT_FORM_UENC 0x80 /* message content type is application/x-www-form-urlencoded */
48990 urbaneks 169
 
54055 urbaneks 170
struct buffer {
171
    struct buffer *next, *prev;
59184 ripley 172
    size_t size, length;
54055 urbaneks 173
    char data[1];
174
};
175
 
59338 urbaneks 176
/* we have to protect re-entrance and not continue processing if there is
177
   a worker inside R already. If we did not then another client connection
178
   would trigger handler and pile up eval on top of the stack, leading to
179
   exhaustion very quickly and a big mess */
180
static int in_process;
181
 
48990 urbaneks 182
/* --- connection/worker structure holding all data for an active connection --- */
183
typedef struct httpd_conn {
184
    SOCKET sock;         /* client socket */
185
    struct in_addr peer; /* IP address of the peer */
186
#ifdef WIN32
187
    HANDLE thread;       /* worker thread */
188
#else
189
    InputHandler *ih;    /* worker input handler */
190
#endif
191
    char line_buf[LINE_BUF_SIZE];  /* line buffer (used for request and headers) */
192
    char *url, *body;              /* URL and request body */
51562 urbaneks 193
    char *content_type;            /* content type (if set) */
59184 ripley 194
    size_t line_pos, body_pos; /* positions in the buffers */
52878 urbaneks 195
    long content_length;           /* desired content length */
48990 urbaneks 196
    char part, method, attr;       /* request part, method and connection attributes */
54055 urbaneks 197
    struct buffer *headers;        /* buffer holding header lines */
48990 urbaneks 198
} httpd_conn_t;
199
 
51562 urbaneks 200
#define IS_HTTP_1_1(C) (((C)->attr & HTTP_1_0) == 0)
201
 
49887 urbaneks 202
/* returns the HTTP/x.x string for a given connection - we support 1.0 and 1.1 only */
51562 urbaneks 203
#define HTTP_SIG(C) (IS_HTTP_1_1(C) ? "HTTP/1.1" : "HTTP/1.0")
49887 urbaneks 204
 
51562 urbaneks 205
/* --- static list of currently active workers --- */
48990 urbaneks 206
static httpd_conn_t *workers[MAX_WORKERS];
207
 
208
/* --- flag determining whether one-time initialization is yet to be performed --- */
209
static int needs_init = 1;
210
 
211
#ifdef WIN32
212
#define WM_RHTTP_CALLBACK ( WM_USER + 1 )
213
static HWND message_window;
49434 ripley 214
static LRESULT CALLBACK
215
RhttpdWindowProc(HWND hwnd, UINT uMsg, WPARAM wParam, LPARAM lParam);
48990 urbaneks 216
#ifndef HWND_MESSAGE
217
#define HWND_MESSAGE ((HWND)-3) /* NOTE: this is supported by W2k/XP and up only! */
218
#endif
219
#endif
220
 
49434 ripley 221
static void first_init()
222
{
48990 urbaneks 223
    initsocks();
224
#ifdef WIN32
49434 ripley 225
    /* create a dummy message-only window for synchronization with the
226
     * main event loop */
48990 urbaneks 227
    HINSTANCE instance = GetModuleHandle(NULL);
228
    LPCTSTR class = "Rhttpd";
49434 ripley 229
    WNDCLASS wndclass = { 0, RhttpdWindowProc, 0, 0, instance, NULL, 0, 0,
230
			  NULL, class };
48990 urbaneks 231
    RegisterClass(&wndclass);
49434 ripley 232
    message_window = CreateWindow(class, "Rhttpd", 0, 1, 1, 1, 1,
233
				  HWND_MESSAGE, NULL, instance, NULL);
48990 urbaneks 234
#endif
235
    needs_init = 0;
236
}
237
 
54055 urbaneks 238
/* free buffers starting from the tail(!!) */
239
static void free_buffer(struct buffer *buf) {
240
    if (!buf) return;
241
    if (buf->prev) free_buffer(buf->prev);
242
    free(buf);
243
}
244
 
245
/* allocate a new buffer */
246
static struct buffer *alloc_buffer(int size, struct buffer *parent) {
247
    struct buffer *buf = (struct buffer*) malloc(sizeof(struct buffer) + size);
248
    if (!buf) return buf;
249
    buf->next = 0;
250
    buf->prev = parent;
251
    if (parent) parent->next = buf;
252
    buf->size = size;
253
    buf->length = 0;
254
    return buf;
255
}
256
 
257
/* convert doubly-linked buffers into one big raw vector */
258
static SEXP collect_buffers(struct buffer *buf) {
259
    SEXP res;
260
    char *dst;
261
    int len = 0;
262
    if (!buf) return allocVector(RAWSXP, 0);
263
    while (buf->prev) { /* count the total length and find the root */
264
	len += buf->length;
265
	buf = buf->prev;
266
    }
267
    res = allocVector(RAWSXP, len + buf->length);
268
    dst = (char*) RAW(res);
269
    while (buf) {
270
	memcpy(dst, buf->data, buf->length);
271
	dst += buf->length;
272
	buf = buf->next;
273
    }
274
    return res;
275
}
276
 
49434 ripley 277
static void finalize_worker(httpd_conn_t *c)
278
{
48990 urbaneks 279
    DBG(printf("finalizing worker %p\n", (void*) c));
280
#ifndef WIN32
281
    if (c->ih) {
282
	removeInputHandler(&R_InputHandlers, c->ih);
283
	c->ih = NULL;
284
    }
285
#endif
286
    if (c->url) {
287
	free(c->url);
288
	c->url = NULL;
289
    }
49434 ripley 290
 
48990 urbaneks 291
    if (c->body) {
292
	free(c->body);
293
	c->body = NULL;
294
    }
49434 ripley 295
 
51562 urbaneks 296
    if (c->content_type) {
297
	free(c->content_type);
298
	c->content_type = NULL;
299
    }
54055 urbaneks 300
    if (c->headers) {
301
	free_buffer(c->headers);
302
	c->headers = NULL;
303
    }
48990 urbaneks 304
    if (c->sock != INVALID_SOCKET) {
305
	closesocket(c->sock);
306
	c->sock = INVALID_SOCKET;
307
    }
308
}
309
 
49434 ripley 310
/* adds a worker to the worker list and returns 0. If the list is
311
 * full, the worker is finalized and returns -1.
312
 * Note that we don't need locking, because add_worker is guaranteed
313
 * to be called by the same thread (server thread).
314
  */
48990 urbaneks 315
static int add_worker(httpd_conn_t *c) {
316
    unsigned int i = 0;
317
    for (; i < MAX_WORKERS; i++)
318
	if (!workers[i]) {
319
#ifdef WIN32
320
	    DBG(printf("registering worker %p as %d (thread=0x%x)\n", (void*) c, i, (int) c->thread));
321
#else
322
	    DBG(printf("registering worker %p as %d (handler=%p)\n", (void*) c, i, (void*) c->ih));
323
#endif
324
	    workers[i] = c;
325
	    return 0;
326
	}
49434 ripley 327
    /* FIXME: ok no more space for a new worker - what do we do now?
328
     * for now we just drop it on the floor .. */
48990 urbaneks 329
    finalize_worker(c);
330
    free(c);
331
    return -1;
332
}
333
 
49434 ripley 334
/* finalize worker, remove it from the list and free the memory. If
335
 * the worker is owned by a thread, it is not finalized and the
336
 * THREAD_DISPOSE flag is set instead. */
337
static void remove_worker(httpd_conn_t *c)
338
{
48990 urbaneks 339
    unsigned int i = 0;
340
    if (!c) return;
49434 ripley 341
    if (c->attr & THREAD_OWNED) { /* if the worker is used by a
342
				   * thread, we can only signal for
343
				   * its removal */
48990 urbaneks 344
	c->attr |= THREAD_DISPOSE;
345
	return;
346
    }
347
    finalize_worker(c);
348
    for (; i < MAX_WORKERS; i++)
349
	if (workers[i] == c)
350
	    workers[i] = NULL;
351
    DBG(printf("removing worker %p\n", (void*) c));
49434 ripley 352
    free(c);
48990 urbaneks 353
}
354
 
51805 urbaneks 355
#ifndef Win32
356
extern int R_ignore_SIGPIPE; /* defined in src/main/main.c on unix */
357
#else
358
static int R_ignore_SIGPIPE; /* for simplicity of the code below */
359
#endif
360
 
59184 ripley 361
static int send_response(SOCKET s, const char *buf, size_t len)
49434 ripley 362
{
48990 urbaneks 363
    unsigned int i = 0;
51805 urbaneks 364
    /* we have to tell R to ignore SIGPIPE otherwise it can raise an error
365
       and get us into deep trouble */
366
    R_ignore_SIGPIPE = 1;
48990 urbaneks 367
    while (i < len) {
59184 ripley 368
	ssize_t n = send(s, buf + i, len - i, 0);
51805 urbaneks 369
	if (n < 1) {
370
	    R_ignore_SIGPIPE = 0;
371
	    return -1;
372
	}
48990 urbaneks 373
	i += n;
374
    }
51805 urbaneks 375
    R_ignore_SIGPIPE = 0;
48990 urbaneks 376
    return 0;
377
}
378
 
49887 urbaneks 379
/* sends HTTP/x.x plus the text (which should be of the form " XXX ...") */
380
static int send_http_response(httpd_conn_t *c, const char *text) {
51562 urbaneks 381
    char buf[96];
49887 urbaneks 382
    const char *s = HTTP_SIG(c);
59184 ripley 383
    size_t l = strlen(text);
384
    ssize_t res;
51562 urbaneks 385
    /* reduce the number of packets by sending the payload en-block from buf */
386
    if (l < sizeof(buf) - 10) {
387
	strcpy(buf, s);
388
	strcpy(buf + 8, text);
389
	return send_response(c->sock, buf, l + 8);
390
    }
51805 urbaneks 391
    R_ignore_SIGPIPE = 1;
392
    res = send(c->sock, s, 8, 0);
393
    R_ignore_SIGPIPE = 0;
394
    if (res < 8) return -1;
49887 urbaneks 395
    return send_response(c->sock, text, strlen(text));
396
}
397
 
48990 urbaneks 398
/* decode URI in place (decoding never expands) */
49434 ripley 399
static void uri_decode(char *s)
400
{
48990 urbaneks 401
    char *t = s;
402
    while (*s) {
403
	if (*s == '+') { /* + -> SPC */
404
	    *(t++) = ' '; s++;
405
	} else if (*s == '%') {
406
	    unsigned char ec = 0;
407
	    s++;
408
	    if (*s >= '0' && *s <= '9') ec |= ((unsigned char)(*s - '0')) << 4;
409
	    else if (*s >= 'a' && *s <= 'f') ec |= ((unsigned char)(*s - 'a' + 10)) << 4;
410
	    else if (*s >= 'A' && *s <= 'F') ec |= ((unsigned char)(*s - 'A' + 10)) << 4;
411
	    if (*s) s++;
412
	    if (*s >= '0' && *s <= '9') ec |= (unsigned char)(*s - '0');
413
	    else if (*s >= 'a' && *s <= 'f') ec |= (unsigned char)(*s - 'a' + 10);
414
	    else if (*s >= 'A' && *s <= 'F') ec |= (unsigned char)(*s - 'A' + 10);
415
	    if (*s) s++;
416
	    *(t++) = (char) ec;
417
	} else *(t++) = *(s++);
418
    }
419
    *t = 0;
420
}
421
 
49434 ripley 422
/* parse a query string into a named character vector - must NOT be
423
 * URI decoded */
424
static SEXP parse_query(char *query)
425
{
48990 urbaneks 426
    int parts = 0;
427
    SEXP res, names;
428
    char *s = query, *key = 0, *value = query, *t = query;
429
    while (*s) {
430
	if (*s == '&') parts++;
431
	s++;
432
    }
433
    parts++;
434
    res = PROTECT(allocVector(STRSXP, parts));
435
    names = PROTECT(allocVector(STRSXP, parts));
436
    s = query;
437
    parts = 0;
438
    while (1) {
439
	if (*s == '=' && !key) { /* first '=' in a part */
440
	    key = value;
441
	    *(t++) = 0;
442
	    value = t;
443
	    s++;
444
	} else if (*s == '&' || !*s) { /* next part */
445
	    int last_entry = !*s;
446
	    *(t++) = 0;
447
	    if (!key) key = "";
448
	    SET_STRING_ELT(names, parts, mkChar(key));
449
	    SET_STRING_ELT(res, parts, mkChar(value));
450
	    parts++;
451
	    if (last_entry) break;
452
	    key = 0;
453
	    value = t;
454
	    s++;
455
	} else if (*s == '+') { /* + -> SPC */
456
	    *(t++) = ' '; s++;
457
	} else if (*s == '%') { /* we cannot use uri_decode becasue we need &/= *before* decoding */
458
	    unsigned char ec = 0;
459
	    s++;
460
	    if (*s >= '0' && *s <= '9') ec |= ((unsigned char)(*s - '0')) << 4;
461
	    else if (*s >= 'a' && *s <= 'f') ec |= ((unsigned char)(*s - 'a' + 10)) << 4;
462
	    else if (*s >= 'A' && *s <= 'F') ec |= ((unsigned char)(*s - 'A' + 10)) << 4;
463
	    if (*s) s++;
464
	    if (*s >= '0' && *s <= '9') ec |= (unsigned char)(*s - '0');
465
	    else if (*s >= 'a' && *s <= 'f') ec |= (unsigned char)(*s - 'a' + 10);
466
	    else if (*s >= 'A' && *s <= 'F') ec |= (unsigned char)(*s - 'A' + 10);
467
	    if (*s) s++;
468
	    *(t++) = (char) ec;
469
	} else *(t++) = *(s++);
470
    }
471
    setAttrib(res, R_NamesSymbol, names);
472
    UNPROTECT(2);
473
    return res;
474
}
475
 
54055 urbaneks 476
static SEXP R_ContentTypeName, R_HandlersName;
477
 
51562 urbaneks 478
/* create an object representing the request body. It is NULL if the body is empty (or zero length).
479
 * In the case of a URL encoded form it will have the same shape as the query string (named string vector).
480
 * In all other cases it will be a raw vector with a "content-type" attribute (if specified in the headers) */
481
static SEXP parse_request_body(httpd_conn_t *c) {
482
    if (!c || !c->body) return R_NilValue;
483
 
484
    if (c->attr & CONTENT_FORM_UENC) { /* URL encoded form - return parsed form */
485
	c->body[c->content_length] = 0; /* the body is guaranteed to have an extra byte for the termination */
486
	return parse_query(c->body);
487
    } else { /* something else - pass it as a raw vector */
488
	SEXP res = PROTECT(Rf_allocVector(RAWSXP, c->content_length));
489
	if (c->content_length)
490
	    memcpy(RAW(res), c->body, c->content_length);
54055 urbaneks 491
	if (c->content_type) { /* attach the content type so it can be interpreted */
492
	    if (!R_ContentTypeName) R_ContentTypeName = install("content-type");
493
	    setAttrib(res, R_ContentTypeName, mkString(c->content_type));
494
	}
51562 urbaneks 495
	UNPROTECT(1);
496
	return res;
497
    }
498
}
499
 
48990 urbaneks 500
#ifdef WIN32
49434 ripley 501
/* on Windows we have to guarantee that process_request is performed
502
 * on the main thread, so we have to dispatch it through a message */
48990 urbaneks 503
static void process_request_main_thread(httpd_conn_t *c);
504
 
49434 ripley 505
static void process_request(httpd_conn_t *c)
506
{
507
    /* SendMessage is synchronous, so it will wait until the message
508
     * is processed */
48990 urbaneks 509
    DBG(Rprintf("enqueuing process_request_main_thread\n"));
510
    SendMessage(message_window, WM_RHTTP_CALLBACK, 0, (LPARAM) c);
511
    DBG(Rprintf("process_request_main_thread returned\n"));
512
}
49434 ripley 513
#define process_request process_request_main_thread
48990 urbaneks 514
#endif
515
 
49887 urbaneks 516
/* finalize a request - essentially for HTTP/1.0 it means that
517
 * we have to close the connection */
518
static void fin_request(httpd_conn_t *c) {
519
    if (!IS_HTTP_1_1(c))
520
	c->attr |= CONNECTION_CLOSE;
521
}
522
 
51563 urbaneks 523
static SEXP custom_handlers_env;
524
 
525
/* returns a httpd handler (closure) for a given path. As a special case
526
 * it can return a symbol that will be resolved in the "tools" namespace.
527
 * currently it allows custom handlers for paths of the form
528
 * /custom/<name>[/.*] where <name> must less than 64 characters long
529
 * and is matched against closures in tools:::.httpd.handlers.env */
530
static SEXP handler_for_path(const char *path) {
531
    if (path && !strncmp(path, "/custom/", 8)) { /* starts with /custom/ ? */
532
	const char *c = path + 8, *e = c;
533
	while (*c && *c != '/') c++; /* find out the name */
534
	if (c - e > 0 && c - e < 64) { /* if it's 1..63 chars long, proceed */
535
	    char fn[64];
536
	    memcpy(fn, e, c - e); /* create a local C string with the name for the install() call */
537
	    fn[c - e] = 0;
538
	    DBG(Rprintf("handler_for_path('%s'): looking up custom handler '%s'\n", path, fn));
539
	    /* we cache custom_handlers_env so in case it has not been loaded yet, fetch it */
54055 urbaneks 540
	    if (!custom_handlers_env) {
541
		if (!R_HandlersName) R_HandlersName = install(".httpd.handlers.env");
542
		custom_handlers_env = eval(R_HandlersName, R_FindNamespace(mkString("tools")));
543
	    }
51563 urbaneks 544
	    /* we only proceed if .httpd.handlers.env really exists */
545
	    if (TYPEOF(custom_handlers_env) == ENVSXP) {
546
		SEXP cl = findVarInFrame3(custom_handlers_env, install(fn), TRUE);
547
		if (cl != R_UnboundValue && TYPEOF(cl) == CLOSXP) /* we need a closure */
548
		    return cl;
549
	    }
550
	}
551
    }
552
    DBG(Rprintf(" - falling back to default httpd\n"));
553
    return install("httpd");
554
}
555
 
48990 urbaneks 556
/* process a request by calling the httpd() function in R */
59338 urbaneks 557
static void process_request_(void *ptr)
49434 ripley 558
{
59338 urbaneks 559
    httpd_conn_t *c = (httpd_conn_t*) ptr;
48990 urbaneks 560
    const char *ct = "text/html";
561
    char *query = 0, *s;
562
    SEXP sHeaders = R_NilValue;
563
    int code = 200;
564
    DBG(Rprintf("process request for %p\n", (void*) c));
565
    if (!c || !c->url) return; /* if there is not enough to process, bail out */
566
    s = c->url;
567
    while (*s && *s != '?') s++; /* find the query part */
568
    if (*s) {
569
	*(s++) = 0;
570
	query = s;
571
    }
572
    uri_decode(c->url); /* decode the path part */
51562 urbaneks 573
    {   /* construct "try(httpd(url, query, body), silent=TRUE)" */
48990 urbaneks 574
	SEXP sTrue = PROTECT(ScalarLogical(TRUE));
54055 urbaneks 575
	SEXP sBody = PROTECT(parse_request_body(c));
576
	SEXP sQuery = PROTECT(query ? parse_query(query) : R_NilValue);
577
	SEXP sReqHeaders = PROTECT(c->headers ? collect_buffers(c->headers) : R_NilValue);
578
	SEXP sArgs = PROTECT(list4(mkString(c->url), sQuery, sBody, sReqHeaders));
579
	SEXP sTry = install("try");
580
	SEXP y, x = PROTECT(lang3(sTry,
581
				  LCONS(handler_for_path(c->url), sArgs),
51562 urbaneks 582
				  sTrue));
48990 urbaneks 583
	SET_TAG(CDR(CDR(x)), install("silent"));
584
	DBG(Rprintf("eval(try(httpd('%s'),silent=TRUE))\n", c->url));
54055 urbaneks 585
 
51562 urbaneks 586
	/* evaluate the above in the tools namespace */
49434 ripley 587
	x = PROTECT(eval(x, R_FindNamespace(mkString("tools"))));
48990 urbaneks 588
 
589
	/* the result is expected to have one of the following forms:
590
 
49434 ripley 591
	   a) character vector of length 1 => error (possibly from try),
592
	      will create 500 response
593
 
594
	  b) list(payload[, content-type[, headers[, status code]]])
595
 
596
	      payload: can be a character vector of length one or a
597
	      raw vector. if the character vector is named "file" then
598
	      the content of a file of that name is the payload
599
 
600
	      content-type: must be a character vector of length one
601
	      or NULL (if present, else default is "text/html")
602
 
603
	      headers: must be a character vector - the elements will
604
	      have CRLF appended and neither Content-type nor
605
	      Content-length may be used
606
 
607
	     status code: must be an integer if present (default is 200)
608
	 */
609
 
48990 urbaneks 610
	if (TYPEOF(x) == STRSXP && LENGTH(x) > 0) { /* string means there was an error */
611
	    const char *s = CHAR(STRING_ELT(x, 0));
49887 urbaneks 612
	    send_http_response(c, " 500 Evaluation error\r\nConnection: close\r\nContent-type: text/plain\r\n\r\n");
48990 urbaneks 613
	    DBG(Rprintf("respond with 500 and content: %s\n", s));
614
	    if (c->method != METHOD_HEAD)
615
		send_response(c->sock, s, strlen(s));
616
	    c->attr |= CONNECTION_CLOSE; /* force close */
54055 urbaneks 617
	    UNPROTECT(7);
48990 urbaneks 618
	    return;
619
	}
48995 urbaneks 620
 
621
	if (TYPEOF(x) == VECSXP && LENGTH(x) > 0) { /* a list (generic vector) can be a real payload */
622
	    SEXP xNames = getAttrib(x, R_NamesSymbol);
48990 urbaneks 623
	    if (LENGTH(x) > 1) {
48995 urbaneks 624
		SEXP sCT = VECTOR_ELT(x, 1); /* second element is content type if present */
48990 urbaneks 625
		if (TYPEOF(sCT) == STRSXP && LENGTH(sCT) > 0)
626
		    ct = CHAR(STRING_ELT(sCT, 0));
48995 urbaneks 627
		if (LENGTH(x) > 2) { /* third element is headers vector */
48990 urbaneks 628
		    sHeaders = VECTOR_ELT(x, 2);
629
		    if (TYPEOF(sHeaders) != STRSXP)
630
			sHeaders = R_NilValue;
48995 urbaneks 631
		    if (LENGTH(x) > 3) /* fourth element is HTTP code */
48990 urbaneks 632
			code = asInteger(VECTOR_ELT(x, 3));
633
		}
634
	    }
635
	    y = VECTOR_ELT(x, 0);
636
	    if (TYPEOF(y) == STRSXP && LENGTH(y) > 0) {
637
		char buf[64];
48995 urbaneks 638
		const char *cs = CHAR(STRING_ELT(y, 0)), *fn = 0;
48990 urbaneks 639
		if (code == 200)
49887 urbaneks 640
		    send_http_response(c, " 200 OK\r\nContent-type: ");
48990 urbaneks 641
		else {
49887 urbaneks 642
		    sprintf(buf, "%s %d Code %d\r\nContent-type: ", HTTP_SIG(c), code, code);
48990 urbaneks 643
		    send_response(c->sock, buf, strlen(buf));
644
		}
645
		send_response(c->sock, ct, strlen(ct));
646
		if (sHeaders != R_NilValue) {
647
		    unsigned int i = 0, n = LENGTH(sHeaders);
648
		    for (; i < n; i++) {
649
			const char *hs = CHAR(STRING_ELT(sHeaders, i));
650
			send_response(c->sock, "\r\n", 2);
651
			send_response(c->sock, hs, strlen(hs));
652
		    }
653
		}
48995 urbaneks 654
		/* special content - a file: either list(file="") or list(c("*FILE*", "")) - the latter will go away */
655
		if (TYPEOF(xNames) == STRSXP && LENGTH(xNames) > 0 &&
656
		    !strcmp(CHAR(STRING_ELT(xNames, 0)), "file"))
657
		    fn = cs;
658
		if (LENGTH(y) > 1 && !strcmp(cs, "*FILE*"))
659
		    fn = CHAR(STRING_ELT(y, 1));
660
		if (fn) {
48990 urbaneks 661
		    char *fbuf;
662
		    FILE *f = fopen(fn, "rb");
663
		    long fsz = 0;
664
		    if (!f) {
665
			send_response(c->sock, "\r\nContent-length: 0\r\n\r\n", 23);
54055 urbaneks 666
			UNPROTECT(7);
49887 urbaneks 667
			fin_request(c);
48990 urbaneks 668
			return;
669
		    }
670
		    fseek(f, 0, SEEK_END);
671
		    fsz = ftell(f);
672
		    fseek(f, 0, SEEK_SET);
673
		    sprintf(buf, "\r\nContent-length: %ld\r\n\r\n", fsz);
48993 urbaneks 674
		    send_response(c->sock, buf, strlen(buf));
48990 urbaneks 675
		    if (c->method != METHOD_HEAD) {
676
			fbuf = (char*) malloc(32768);
51562 urbaneks 677
			if (fbuf) {
678
			    while (fsz > 0 && !feof(f)) {
59184 ripley 679
				size_t rd = (fsz > 32768) ? 32768 : fsz;
51562 urbaneks 680
				if (fread(fbuf, 1, rd, f) != rd) {
681
				    free(fbuf);
54055 urbaneks 682
				    UNPROTECT(7);
51562 urbaneks 683
				    c->attr |= CONNECTION_CLOSE;
684
				    return;
685
				}
686
				send_response(c->sock, fbuf, rd);
687
				fsz -= rd;
48990 urbaneks 688
			    }
51562 urbaneks 689
			    free(fbuf);
690
			} else { /* allocation error - get out */
54055 urbaneks 691
			    UNPROTECT(7);
51562 urbaneks 692
			    c->attr |= CONNECTION_CLOSE;
693
			    return;
48990 urbaneks 694
			}
695
		    }
696
		    fclose(f);
54055 urbaneks 697
		    UNPROTECT(7);
49887 urbaneks 698
		    fin_request(c);
48990 urbaneks 699
		    return;
700
		}
701
		sprintf(buf, "\r\nContent-length: %u\r\n\r\n", (unsigned int) strlen(cs));
702
		send_response(c->sock, buf, strlen(buf));
703
		if (c->method != METHOD_HEAD)
704
		    send_response(c->sock, cs, strlen(cs));
54055 urbaneks 705
		UNPROTECT(7);
49887 urbaneks 706
		fin_request(c);
48990 urbaneks 707
		return;
708
	    }
709
	    if (TYPEOF(y) == RAWSXP) {
710
		char buf[64];
711
		Rbyte *cs = RAW(y);
712
		if (code == 200)
49887 urbaneks 713
		    send_http_response(c, " 200 OK\r\nContent-type: ");
48990 urbaneks 714
		else {
49887 urbaneks 715
		    sprintf(buf, "%s %d Code %d\r\nContent-type: ", HTTP_SIG(c), code, code);
48990 urbaneks 716
		    send_response(c->sock, buf, strlen(buf));
717
		}
718
		send_response(c->sock, ct, strlen(ct));
719
		if (sHeaders != R_NilValue) {
720
		    unsigned int i = 0, n = LENGTH(sHeaders);
721
		    for (; i < n; i++) {
722
			const char *hs = CHAR(STRING_ELT(sHeaders, i));
723
			send_response(c->sock, "\r\n", 2);
724
			send_response(c->sock, hs, strlen(hs));
725
		    }
726
		}
727
		sprintf(buf, "\r\nContent-length: %u\r\n\r\n", LENGTH(y));
728
		send_response(c->sock, buf, strlen(buf));
729
		if (c->method != METHOD_HEAD)
730
		    send_response(c->sock, (char*) cs, LENGTH(y));
54055 urbaneks 731
		UNPROTECT(7);
49887 urbaneks 732
		fin_request(c);
48990 urbaneks 733
		return;
734
	    }
735
	}
54055 urbaneks 736
	UNPROTECT(7);
48990 urbaneks 737
    }
49887 urbaneks 738
    send_http_response(c, " 500 Invalid response from R\r\nConnection: close\r\nContent-type: text/plain\r\n\r\nServer error: invalid response from R\r\n");
48990 urbaneks 739
    c->attr |= CONNECTION_CLOSE; /* force close */
740
}
741
 
59338 urbaneks 742
/* wrap the actual call with ToplevelExec since we need to have a guaranteed
743
   return so we can track the presence of a worker code inside R to prevent
744
   re-entrance from other clients */
745
static void process_request(httpd_conn_t *c)
746
{
747
    in_process = 1;
748
    R_ToplevelExec(process_request_, c);
749
    in_process = 0;
750
}
751
 
48990 urbaneks 752
#ifdef WIN32
753
#undef process_request
754
#endif
755
 
49434 ripley 756
/* this function is called to fetch new data from the client
757
 * connection socket and process it */
48990 urbaneks 758
static void worker_input_handler(void *data) {
759
    httpd_conn_t *c = (httpd_conn_t*) data;
49434 ripley 760
 
48990 urbaneks 761
    DBG(printf("worker_input_handler, data=%p\n", data));
762
    if (!c) return;
49434 ripley 763
 
59338 urbaneks 764
    if (in_process) return; /* we don't allow recursive entrance */
765
 
48990 urbaneks 766
    DBG(printf("input handler for worker %p (sock=%d, part=%d, method=%d, line_pos=%d)\n", (void*) c, (int)c->sock, (int)c->part, (int)c->method, (int)c->line_pos));
49434 ripley 767
 
768
    /* FIXME: there is one edge case that is not caught on unix: if
769
     * recv reads two or more full requests into the line buffer then
770
     * this function exits after the first one, but input handlers may
771
     * not trigger, because there may be no further data. It is not
772
     * trivial to fix, because just checking for a full line at the
773
     * beginning and not calling recv won't trigger a new input
774
     * handler. However, under normal circumstance this should not
775
     * happen, because clients should wait for the response and even
776
     * if they don't it's unlikely that both requests get combined
777
     * into one packet. */
48990 urbaneks 778
    if (c->part < PART_BODY) {
779
	char *s = c->line_buf;
59184 ripley 780
	ssize_t n = recv(c->sock, c->line_buf + c->line_pos, 
781
			 LINE_BUF_SIZE - c->line_pos - 1, 0);
48990 urbaneks 782
	DBG(printf("[recv n=%d, line_pos=%d, part=%d]\n", n, c->line_pos, (int)c->part));
783
	if (n < 0) { /* error, scrape this worker */
784
	    remove_worker(c);
785
	    return;
786
	}
787
	if (n == 0) { /* connection closed -> try to process and then remove */
788
	    process_request(c);
789
	    remove_worker(c);
790
	    return;
791
	}
792
	c->line_pos += n;
793
	c->line_buf[c->line_pos] = 0;
794
	DBG(printf("in buffer: {%s}\n", c->line_buf));
795
	while (*s) {
796
	    /* ok, we have genuine data in the line buffer */
797
	    if (s[0] == '\n' || (s[0] == '\r' && s[1] == '\n')) { /* single, empty line - end of headers */
798
		/* --- check request validity --- */
799
		DBG(printf(" end of request, moving to body\n"));
800
		if (!(c->attr & HTTP_1_0) && !(c->attr & HOST_HEADER)) { /* HTTP/1.1 mandates Host: header */
49887 urbaneks 801
		    send_http_response(c, " 400 Bad Request (Host: missing)\r\nConnection: close\r\n\r\n");
48990 urbaneks 802
		    remove_worker(c);
803
		    return;
804
		}
51562 urbaneks 805
		if (c->attr & CONTENT_LENGTH && c->content_length) {
52878 urbaneks 806
		    if (c->content_length < 0 ||  /* we are parsing signed so negative numbers are bad */
807
			c->content_length > 2147483640 || /* R will currently have issues with body around 2Gb or more, so better to not go there */
808
			!(c->body = (char*) malloc(c->content_length + 1 /* allocate an extra termination byte */ ))) {
51562 urbaneks 809
			send_http_response(c, " 413 Request Entity Too Large (request body too big)\r\nConnection: close\r\n\r\n");
810
			remove_worker(c);
811
			return;
812
		    }
48990 urbaneks 813
		}
814
		c->body_pos = 0;
815
		c->part = PART_BODY;
816
		if (s[0] == '\r') s++;
817
		s++;
818
		/* move the body part to the beginning of the buffer */
819
		c->line_pos -= s - c->line_buf;
820
		memmove(c->line_buf, s, c->line_pos);
62724 urbaneks 821
		/* GET/HEAD or no content length mean no body */
822
		if (c->method == METHOD_GET || c->method == METHOD_HEAD ||
823
		    !(c->attr & CONTENT_LENGTH) || c->content_length == 0) {
62770 urbaneks 824
		    if ((c->attr & CONTENT_LENGTH) && c->content_length > 0) {
51805 urbaneks 825
			send_http_response(c, " 400 Bad Request (GET/HEAD with body)\r\n\r\n");
48990 urbaneks 826
			remove_worker(c);
827
			return;
828
		    }
829
		    process_request(c);
830
		    if (c->attr & CONNECTION_CLOSE) {
831
			remove_worker(c);
832
			return;
833
		    }
834
		    /* keep-alive - reset the worker so it can process a new request */
835
		    if (c->url) { free(c->url); c->url = NULL; }
836
		    if (c->body) { free(c->body); c->body = NULL; }
51562 urbaneks 837
		    if (c->content_type) { free(c->content_type); c->content_type = NULL; }
54055 urbaneks 838
		    if (c->headers) { free_buffer(c->headers); c->headers = NULL; }
48990 urbaneks 839
		    c->body_pos = 0;
840
		    c->method = 0;
841
		    c->part = PART_REQUEST;
842
		    c->attr = 0;
843
		    c->content_length = 0;
844
		    return;
845
		}
51653 urbaneks 846
		/* copy body content (as far as available) */
847
		c->body_pos = (c->content_length < c->line_pos) ? c->content_length : c->line_pos;
848
		if (c->body_pos) {
849
		    memcpy(c->body, c->line_buf, c->body_pos);
850
		    c->line_pos -= c->body_pos; /* NOTE: we are NOT moving the buffer since non-zero left-over causes connection close */
851
		}
48990 urbaneks 852
		/* POST will continue into the BODY part */
853
		break;
854
	    }
855
	    {
856
		char *bol = s;
857
		while (*s && *s != '\r' && *s != '\n') s++;
858
		if (!*s) { /* incomplete line */
859
		    if (bol == c->line_buf) {
860
			if (c->line_pos < LINE_BUF_SIZE) /* one, incomplete line, but the buffer is not full yet, just return */
861
			    return;
862
			/* the buffer is full yet the line is incomplete - we're in trouble */
51805 urbaneks 863
			send_http_response(c, " 413 Request entity too large\r\nConnection: close\r\n\r\n");
48990 urbaneks 864
			remove_worker(c);
865
			return;
866
		    }
867
		    /* move the line to the begining of the buffer for later requests */
868
		    c->line_pos -= bol - c->line_buf;
869
		    memmove(c->line_buf, bol, c->line_pos);
870
		    return;
871
		} else { /* complete line, great! */
872
		    if (*s == '\r') *(s++) = 0;
873
		    if (*s == '\n') *(s++) = 0;
874
		    DBG(printf("complete line: {%s}\n", bol));
875
		    if (c->part == PART_REQUEST) {
876
			/* --- process request line --- */
59184 ripley 877
			size_t rll = strlen(bol); /* request line length */
62724 urbaneks 878
			char *url = strchr(bol, ' ');
879
			if (!url || rll < 14 || strncmp(bol + rll - 9, " HTTP/1.", 8)) { /* each request must have at least 14 characters [GET / HTTP/1.0] and have HTTP/1.x */
51805 urbaneks 880
			    send_response(c->sock, "HTTP/1.0 400 Bad Request\r\n\r\n", 28);
48990 urbaneks 881
			    remove_worker(c);
882
			    return;
883
			}
62724 urbaneks 884
			url++;
48990 urbaneks 885
			if (!strncmp(bol + rll - 3, "1.0", 3)) c->attr |= HTTP_1_0;
62724 urbaneks 886
			if (!strncmp(bol, "GET ", 4)) c->method = METHOD_GET;
48990 urbaneks 887
			if (!strncmp(bol, "POST ", 5)) c->method = METHOD_POST;
888
			if (!strncmp(bol, "HEAD ", 5)) c->method = METHOD_HEAD;
62724 urbaneks 889
			/* only custom handlers can use other methods */
62726 urbaneks 890
			if (!strncmp(url, "/custom/", 8)) {
62724 urbaneks 891
			    char *mend = url - 1;
892
			    /* we generate a header with the method so it can be passed to the handler */
893
			    if (!c->headers)
894
				c->headers = alloc_buffer(1024, NULL);
895
			    /* make sure it fits */
896
			    if (c->headers->size - c->headers->length >= 18 + (mend - bol)) {
62726 urbaneks 897
				if (!c->method) c->method = METHOD_OTHER;
62724 urbaneks 898
				/* add "Request-Method: xxx" */
899
				memcpy(c->headers->data + c->headers->length, "Request-Method: ", 16);
900
				c->headers->length += 16;
901
				memcpy(c->headers->data + c->headers->length, bol, mend - bol);
902
				c->headers->length += mend - bol;	
903
				c->headers->data[c->headers->length++] = '\n';
904
			    }
905
			}
48990 urbaneks 906
			if (!c->method) {
51805 urbaneks 907
			    send_http_response(c, " 501 Invalid or unimplemented method\r\n\r\n");
48990 urbaneks 908
			    remove_worker(c);
909
			    return;
910
			}
911
			bol[strlen(bol) - 9] = 0;
912
			c->url = strdup(url);
913
			c->part = PART_HEADER;
914
			DBG(printf("parsed request, method=%d, URL='%s'\n", (int)c->method, c->url));
915
		    } else if (c->part == PART_HEADER) {
916
			/* --- process headers --- */
917
			char *k = bol;
54055 urbaneks 918
			if (!c->headers)
919
			    c->headers = alloc_buffer(1024, NULL);
920
			if (c->headers) { /* record the header line in the buffer */
59184 ripley 921
			    size_t l = strlen(bol);
54055 urbaneks 922
			    if (l) { /* this should be really always true */
923
				if (c->headers->length + l + 1 > c->headers->size) { /* not enough space? */
59184 ripley 924
				    size_t fits = c->headers->size - c->headers->length;
54055 urbaneks 925
				    if (fits) memcpy(c->headers->data + c->headers->length, bol, fits);
926
				    if (alloc_buffer(2048, c->headers)) {
927
					c->headers = c->headers->next;
54327 urbaneks 928
					memcpy(c->headers->data, bol + fits, l - fits);
54055 urbaneks 929
					c->headers->length = l - fits;
930
					c->headers->data[c->headers->length++] = '\n';
931
				    }
932
				} else {
933
				    memcpy(c->headers->data + c->headers->length, bol, l);
934
				    c->headers->length += l;	
935
				    c->headers->data[c->headers->length++] = '\n';
936
				}
937
			    }
938
			}
48990 urbaneks 939
			while (*k && *k != ':') {
940
			    if (*k >= 'A' && *k <= 'Z')
941
				*k |= 0x20;
942
			    k++;
943
			}
944
			if (*k == ':') {
945
			    *(k++) = 0;
946
			    while (*k == ' ' || *k == '\t') k++;
947
			    DBG(printf("header '%s' => '%s'\n", bol, k));
948
			    if (!strcmp(bol, "content-length")) {
949
				c->attr |= CONTENT_LENGTH;
52878 urbaneks 950
				c->content_length = atol(k);
48990 urbaneks 951
			    }
51562 urbaneks 952
			    if (!strcmp(bol, "content-type")) {
953
				char *l = k;
954
				while (*l) { if (*l >= 'A' && *l <= 'Z') *l |= 0x20; l++; }
955
				c->attr |= CONTENT_TYPE;
956
				if (c->content_type) free(c->content_type);
957
				c->content_type = strdup(k);
958
				if (!strncmp(k, "application/x-www-form-urlencoded", 33))
959
				    c->attr |= CONTENT_FORM_UENC;
960
			    }
48990 urbaneks 961
			    if (!strcmp(bol, "host"))
962
				c->attr |= HOST_HEADER;
963
			    if (!strcmp(bol, "connection")) {
964
				char *l = k;
965
				while (*l) { if (*l >= 'A' && *l <= 'Z') *l |= 0x20; l++; }
966
				if (!strncmp(k, "close", 5))
54979 urbaneks 967
				    c->attr |= CONNECTION_CLOSE;
48990 urbaneks 968
			    }
969
			}
970
		    }
971
		}
972
	    }
973
	}
51653 urbaneks 974
	if (c->part < PART_BODY) {
975
	    /* we end here if we processed a buffer of exactly one line */
976
	    c->line_pos = 0;
48990 urbaneks 977
	    return;
978
	}
51653 urbaneks 979
    }
980
    if (c->part == PART_BODY && c->body) { /* BODY  - this branch always returns */
981
	if (c->body_pos < c->content_length) { /* need to receive more ? */
52878 urbaneks 982
	    DBG(printf("BODY: body_pos=%d, content_length=%ld\n", c->body_pos, c->content_length));
59184 ripley 983
	    ssize_t n = recv(c->sock, c->body + c->body_pos, 
984
			    c->content_length - c->body_pos, 0);
52878 urbaneks 985
	    DBG(printf("      [recv n=%d - had %u of %lu]\n", n, c->body_pos, c->content_length));
51653 urbaneks 986
	    c->line_pos = 0;
987
	    if (n < 0) { /* error, scrap this worker */
988
		remove_worker(c);
989
		return;
990
	    }
991
	    if (n == 0) { /* connection closed -> try to process and then remove */
992
		process_request(c);
48990 urbaneks 993
	    remove_worker(c);
51653 urbaneks 994
		return;
995
	    }
996
	    c->body_pos += n;
48990 urbaneks 997
	}
998
	if (c->body_pos == c->content_length) { /* yay! we got the whole body */
999
	    process_request(c);
51653 urbaneks 1000
	    if (c->attr & CONNECTION_CLOSE || c->line_pos) { /* we have to close the connection if there was a double-hit */
48990 urbaneks 1001
		remove_worker(c);
1002
		return;
1003
	    }
1004
	    /* keep-alive - reset the worker so it can process a new request */
1005
	    if (c->url) { free(c->url); c->url = NULL; }
1006
	    if (c->body) { free(c->body); c->body = NULL; }
51562 urbaneks 1007
	    if (c->content_type) { free(c->content_type); c->content_type = NULL; }
54327 urbaneks 1008
	    if (c->headers) { free_buffer(c->headers); c->headers = NULL; }
48990 urbaneks 1009
	    c->line_pos = 0; c->body_pos = 0;
1010
	    c->method = 0;
1011
	    c->part = PART_REQUEST;
1012
	    c->attr = 0;
1013
	    c->content_length = 0;
1014
	    return;
1015
	}
1016
    }
49434 ripley 1017
 
48990 urbaneks 1018
    /* we enter here only if recv was used to leave the headers with no body */
1019
    if (c->part == PART_BODY && !c->body) {
1020
	char *s = c->line_buf;
1021
	if (c->line_pos > 0) {
1022
	    if ((s[0] != '\r' || s[1] != '\n') && (s[0] != '\n')) {
49887 urbaneks 1023
		send_http_response(c, " 411 length is required for non-empty body\r\nConnection: close\r\n\r\n");
48990 urbaneks 1024
		remove_worker(c);
1025
		return;
1026
	    }
1027
	    /* empty body, good */
1028
	    process_request(c);
1029
	    if (c->attr & CONNECTION_CLOSE) {
1030
		remove_worker(c);
1031
		return;
1032
	    } else { /* keep-alive */
1033
		int sh = 1;
1034
		if (s[0] == '\r') sh++;
1035
		if (c->line_pos <= sh)
1036
		    c->line_pos = 0;
1037
		else { /* shift the remaining buffer */
1038
		    memmove(c->line_buf, c->line_buf + sh, c->line_pos - sh);
1039
		    c->line_pos -= sh;
1040
		}
1041
		/* keep-alive - reset the worker so it can process a new request */
1042
		if (c->url) { free(c->url); c->url = NULL; }
1043
		if (c->body) { free(c->body); c->body = NULL; }
51562 urbaneks 1044
		if (c->content_type) { free(c->content_type); c->content_type = NULL; }
54327 urbaneks 1045
		if (c->headers) { free_buffer(c->headers); c->headers = NULL; }
48990 urbaneks 1046
		c->body_pos = 0;
1047
		c->method = 0;
1048
		c->part = PART_REQUEST;
1049
		c->attr = 0;
1050
		c->content_length = 0;
1051
		return;
1052
	    }
1053
	}
59184 ripley 1054
	ssize_t n = recv(c->sock, c->line_buf + c->line_pos, 
1055
			 LINE_BUF_SIZE - c->line_pos - 1, 0);
51562 urbaneks 1056
	if (n < 0) { /* error, scrap this worker */
48990 urbaneks 1057
	    remove_worker(c);
1058
	    return;
1059
	}
1060
	if (n == 0) { /* connection closed -> try to process and then remove */
1061
	    process_request(c);
1062
	    remove_worker(c);
1063
	    return;
1064
	}
1065
	if ((s[0] != '\r' || s[1] != '\n') && (s[0] != '\n')) {
49887 urbaneks 1066
	    send_http_response(c, " 411 length is required for non-empty body\r\nConnection: close\r\n\r\n");
48990 urbaneks 1067
	    remove_worker(c);
1068
	    return;
49434 ripley 1069
	}
48990 urbaneks 1070
    }
1071
}
1072
 
1073
static void srv_input_handler(void *data);
1074
 
1075
static SOCKET srv_sock = INVALID_SOCKET;
1076
 
1077
#ifdef WIN32
49434 ripley 1078
/* Windows implementation uses threads to accept and serve
1079
   connections, using the main event loop to synchronize with R
1080
   through a message-only window which is created on the R thread
1081
 */
48990 urbaneks 1082
static LRESULT CALLBACK RhttpdWindowProc(HWND hwnd, UINT uMsg, WPARAM wParam, LPARAM lParam)
1083
{
48992 urbaneks 1084
    DBG(Rprintf("RhttpdWindowProc(%x, %x, %x, %x)\n", (int) hwnd, (int) uMsg, (int) wParam, (int) lParam));
48990 urbaneks 1085
    if (hwnd == message_window && uMsg == WM_RHTTP_CALLBACK) {
1086
	httpd_conn_t *c = (httpd_conn_t*) lParam;
1087
	process_request_main_thread(c);
1088
	return 0;
1089
    }
1090
    return DefWindowProc(hwnd, uMsg, wParam, lParam);
1091
}
1092
 
49434 ripley 1093
/* server thread - accepts connections on the server socket and
1094
   creates worker threads
1095
 */
48990 urbaneks 1096
static DWORD WINAPI ServerThreadProc(LPVOID lpParameter) {
1097
    while (srv_sock != INVALID_SOCKET) {
1098
	srv_input_handler(lpParameter);
1099
    }
1100
    return 0;
1101
}
1102
 
1103
/* worker thread - processes one client connection socket */
1104
static DWORD WINAPI WorkerThreadProc(LPVOID lpParameter) {
1105
    httpd_conn_t *c = (httpd_conn_t*) lpParameter;
1106
    if (!c) return 0;
1107
    while ((c->attr & THREAD_DISPOSE) == 0) {
1108
	c->attr |= THREAD_OWNED; /* make sure the worker is not removed by the handler since we need it */
1109
	worker_input_handler(c);
1110
    }
1111
    /* the handler signalled a desire to remove the worker, do it */
1112
    c->attr = 0; /* reset the flags */
1113
    remove_worker(c); /* free the worker */
1114
    return 0;
1115
}
1116
 
1117
/* global server thread - currently we support only one server at a time */
1118
HANDLE server_thread;
1119
#else
49434 ripley 1120
/* on unix we register all used sockets (server and workers) as input
1121
 * handlers such that we can avoid polling */
48990 urbaneks 1122
 
1123
/* global input handler for the server socket */
1124
static InputHandler *srv_handler;
1125
#endif
1126
 
49434 ripley 1127
static void srv_input_handler(void *data)
1128
{
48990 urbaneks 1129
    httpd_conn_t *c;
1130
    SAIN peer_sa;
1131
    socklen_t peer_sal = sizeof(peer_sa);
1132
    SOCKET cl_sock = accept(srv_sock, (SA*) &peer_sa, &peer_sal);
1133
    if (cl_sock == INVALID_SOCKET) /* accept failed, don't bother */
1134
	return;
1135
    c = (httpd_conn_t*) calloc(1, sizeof(httpd_conn_t));
1136
    c->sock = cl_sock;
1137
    c->peer = peer_sa.sin_addr;
1138
#ifndef WIN32
49434 ripley 1139
    c->ih = addInputHandler(R_InputHandlers, cl_sock, &worker_input_handler,
1140
			    HttpdWorkerActivity);
48990 urbaneks 1141
    if (c->ih) c->ih->userData = c;
1142
    add_worker(c);
1143
#else
49434 ripley 1144
    if (!add_worker(c)) { /* create worker thread only if the worker
1145
			   * was accepted */
1146
	if (!(c->thread = CreateThread(NULL, 0, WorkerThreadProc,
1147
				       (LPVOID) c, 0, 0)))
48990 urbaneks 1148
	    remove_worker(c);
1149
    }
1150
#endif
1151
}
1152
 
49434 ripley 1153
int in_R_HTTPDCreate(const char *ip, int port) 
1154
{
49788 ripley 1155
#ifndef WIN32
48990 urbaneks 1156
    int reuse = 1;
49788 ripley 1157
#endif
48990 urbaneks 1158
    SAIN srv_sa;
49434 ripley 1159
 
48990 urbaneks 1160
    if (needs_init) /* initialization may need to be performed on first use */
1161
	first_init();
49434 ripley 1162
 
48990 urbaneks 1163
    /* is already in use, close the current socket */
1164
    if (srv_sock != INVALID_SOCKET)
1165
	closesocket(srv_sock);
49434 ripley 1166
 
48990 urbaneks 1167
#ifdef WIN32
1168
    /* on Windows stop the server thread if it exists */
1169
    if (server_thread) {
1170
	DWORD ts = 0;
1171
	if (GetExitCodeThread(server_thread, &ts) && ts == STILL_ACTIVE)
1172
	    TerminateThread(server_thread, 0);
1173
	server_thread = 0;
1174
    }
49434 ripley 1175
#endif
1176
 
48990 urbaneks 1177
    /* create a new socket */
1178
    srv_sock = socket(AF_INET, SOCK_STREAM, 0);
1179
    if (srv_sock == INVALID_SOCKET)
1180
	Rf_error("unable to create socket");
49434 ripley 1181
 
49754 murdoch 1182
#ifndef WIN32
48990 urbaneks 1183
    /* set socket for reuse so we can re-init if we die */
49754 murdoch 1184
    /* But on Windows, this lets us stomp on any port already in use, so don't do it. */
49434 ripley 1185
    setsockopt(srv_sock, SOL_SOCKET, SO_REUSEADDR,
1186
	       (const char*)&reuse, sizeof(reuse));
49754 murdoch 1187
#endif
49434 ripley 1188
 
48990 urbaneks 1189
    /* bind to the desired port */
1190
    if (bind(srv_sock, build_sin(&srv_sa, ip, port), sizeof(srv_sa))) {
1191
	if (sockerrno == EADDRINUSE) {
1192
	    closesocket(srv_sock);
1193
	    srv_sock = INVALID_SOCKET;
1194
	    return -2;
1195
	} else {
1196
	    closesocket(srv_sock);
1197
	    srv_sock = INVALID_SOCKET;
1198
	    Rf_error("unable to bind socket to TCP port %d", port);
1199
	}
1200
    }
49434 ripley 1201
 
48990 urbaneks 1202
    /* setup listen */
1203
    if (listen(srv_sock, 8))
1204
	Rf_error("cannot listen to TCP port %d", port);
49434 ripley 1205
 
48990 urbaneks 1206
#ifndef WIN32
1207
    /* all went well, register the socket as a handler */
1208
    if (srv_handler) removeInputHandler(&R_InputHandlers, srv_handler);
49434 ripley 1209
    srv_handler = addInputHandler(R_InputHandlers, srv_sock,
1210
				  &srv_input_handler, HttpdServerActivity);
48990 urbaneks 1211
#else
1212
    /* do the desired Windows synchronization */
1213
    server_thread = CreateThread(NULL, 0, ServerThreadProc, 0, 0, 0);
1214
#endif
1215
    return 0;
1216
}
1217
 
49434 ripley 1218
void in_R_HTTPDStop(void)
1219
{
1220
    if (srv_sock != INVALID_SOCKET) closesocket(srv_sock);
1221
    srv_sock = INVALID_SOCKET;
1222
 
1223
#ifdef WIN32
1224
    /* on Windows stop the server thread if it exists */
1225
    if (server_thread) {
1226
	DWORD ts = 0;
1227
	if (GetExitCodeThread(server_thread, &ts) && ts == STILL_ACTIVE)
1228
	    TerminateThread(server_thread, 0);
1229
	server_thread = 0;
1230
    }
1231
#else
1232
    if (srv_handler) removeInputHandler(&R_InputHandlers, srv_handler);
1233
#endif
1234
}
1235
 
1236
/* Create an internal http server in R. Note that currently there can
1237
   only be at most one http server running at any given time so the
1238
   behavior is undefined if a server already exists (currently any
1239
   previous servers will be shut down by this call but the shutdown
1240
   may not be clean).
1241
 
1242
   @param sIP is the IP to bind to (or NULL for any)
1243
   @param sPort is the TCP port number to bin to
1244
   @return returns an integer value -- 0L on success, other values
1245
   denote failures: -2L means that the address/port combination is
1246
   already in use
1247
*/
1248
SEXP R_init_httpd(SEXP sIP, SEXP sPort)
1249
{
48990 urbaneks 1250
    const char *ip = 0;
1251
    if (sIP != R_NilValue && (TYPEOF(sIP) != STRSXP || LENGTH(sIP) != 1))
1252
	Rf_error("invalid bind address specification");
1253
    if (sIP != R_NilValue)
1254
	ip = CHAR(STRING_ELT(sIP, 0));
1255
    return ScalarInteger(in_R_HTTPDCreate(ip, asInteger(sPort)));
1256
}