The R Project SVN R

Rev

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