← Back to branch summary

~j-pureftpd/gearmand/tokyo

« back to all changes in this revision

Viewing changes to libgearman/protocol_http.c

  • Committer: Frank Denis
  • Date: 2009-06-22 14:50:52 UTC
  • mfrom: (50.1.1 gearmand)
  • Revision ID: j@jedi.devteam.orbus.fr-20090622145052-s40f87uem2zrmvdz
Sync with trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
libgearman/protocol_http.c
 
1
/* Gearman server and library
 
2
 * Copyright (C) 2008 Brian Aker, Eric Day
 
3
 * All rights reserved.
 
4
 *
 
5
 * Use and distribution licensed under the BSD license.  See
 
6
 * the COPYING file in the parent directory for full text.
 
7
 */
 
8
 
 
9
/**
 
10
 * @file
 
11
 * @brief HTTP Protocol Definitions
 
12
 */
 
13
 
 
14
#include "common.h"
 
15
 
 
16
#include <libgearman/protocol_http.h>
 
17
 
 
18
/**
 
19
 * @addtogroup gearman_protocol_http HTTP Protocol Functions
 
20
 * @ingroup gearman_protocol
 
21
 * @{
 
22
 */
 
23
 
 
24
/**
 
25
 * Default values.
 
26
 */
 
27
#define GEARMAN_PROTOCOL_HTTP_DEFAULT_PORT 8080
 
28
 
 
29
/*
 
30
 * Private declarations
 
31
 */
 
32
 
 
33
/**
 
34
 * Structure for HTTP specific data.
 
35
 */
 
36
typedef struct
 
37
{
 
38
  bool background;
 
39
  bool keep_alive;
 
40
} gearman_protocol_http_st;
 
41
 
 
42
/* Protocol callback functions. */
 
43
static gearman_return_t _http_con_add(gearman_con_st *con);
 
44
static void _http_free(gearman_con_st *con , void *data);
 
45
static size_t _http_pack(gearman_packet_st *packet, gearman_con_st *con,
 
46
                         void *data, size_t data_size,
 
47
                         gearman_return_t *ret_ptr);
 
48
static size_t _http_unpack(gearman_packet_st *packet, gearman_con_st *con,
 
49
                           const void *data, size_t data_size,
 
50
                           gearman_return_t *ret_ptr);
 
51
 
 
52
/* Line parsing helper function. */
 
53
static const char *_http_line(const void *data, size_t data_size,
 
54
                              size_t *line_size, size_t *offset);
 
55
 
 
56
/*
 
57
 * Public definitions
 
58
 */
 
59
 
 
60
gearman_return_t gearman_protocol_http_conf(gearman_conf_st *conf)
 
61
{
 
62
  gearman_conf_module_st *module;
 
63
 
 
64
  module= gearman_conf_module_create(conf, NULL, "http");
 
65
  if (module == NULL)
 
66
    return GEARMAN_MEMORY_ALLOCATION_FAILURE;
 
67
 
 
68
  gearman_conf_module_add_option(module, "port", 0, "PORT",
 
69
                                 "Port to listen on.");
 
70
 
 
71
  return gearman_conf_return(conf);
 
72
}
 
73
 
 
74
gearman_return_t gearmand_protocol_http_init(gearmand_st *gearmand,
 
75
                                             gearman_conf_st *conf)
 
76
{
 
77
  in_port_t port= GEARMAN_PROTOCOL_HTTP_DEFAULT_PORT;
 
78
  gearman_conf_module_st *module;
 
79
  const char *name;
 
80
  const char *value;
 
81
 
 
82
  GEARMAN_INFO(gearmand, "Initializing http module")
 
83
 
 
84
  /* Get module and parse the option values that were given. */
 
85
  module= gearman_conf_module_find(conf, "http");
 
86
  if (module == NULL)
 
87
  {
 
88
    GEARMAN_FATAL(gearmand,
 
89
                  "gearman_protocol_http_init:gearman_conf_module_find:NULL")
 
90
    return GEARMAN_QUEUE_ERROR;
 
91
  }
 
92
 
 
93
  while (gearman_conf_module_value(module, &name, &value))
 
94
  {
 
95
    if (!strcmp(name, "port"))
 
96
      port= (in_port_t)atoi(value);
 
97
    else
 
98
    {
 
99
      gearmand_protocol_http_deinit(gearmand);
 
100
      GEARMAN_FATAL(gearmand, "gearman_protocol_http_init:Unknown argument: %s",
 
101
                    name)
 
102
      return GEARMAN_QUEUE_ERROR;
 
103
    }
 
104
  }
 
105
 
 
106
  return gearmand_port_add(gearmand, port, _http_con_add);
 
107
}
 
108
 
 
109
gearman_return_t gearmand_protocol_http_deinit(gearmand_st *gearmand __attribute__ ((unused)))
 
110
{
 
111
  return GEARMAN_SUCCESS;
 
112
}
 
113
 
 
114
/*
 
115
 * Private definitions
 
116
 */
 
117
 
 
118
static gearman_return_t _http_con_add(gearman_con_st *con)
 
119
{
 
120
  gearman_protocol_http_st *http;
 
121
 
 
122
  http= malloc(sizeof(gearman_protocol_http_st));
 
123
  if (http == NULL)
 
124
  {
 
125
    GEARMAN_ERROR_SET(con->gearman, "_http_con_add", "malloc")
 
126
    return GEARMAN_MEMORY_ALLOCATION_FAILURE;
 
127
  }
 
128
 
 
129
  http->background= false;
 
130
  http->keep_alive= false;
 
131
 
 
132
  gearman_con_set_protocol_data(con, http);
 
133
  gearman_con_set_protocol_data_free_fn(con, _http_free);
 
134
  gearman_con_set_packet_pack_fn(con, _http_pack);
 
135
  gearman_con_set_packet_unpack_fn(con, _http_unpack);
 
136
 
 
137
  return GEARMAN_SUCCESS;
 
138
}
 
139
 
 
140
static void _http_free(gearman_con_st *con __attribute__ ((unused)), void *data)
 
141
{
 
142
  free(data);
 
143
}
 
144
 
 
145
static size_t _http_pack(gearman_packet_st *packet, gearman_con_st *con,
 
146
                         void *data, size_t data_size,
 
147
                         gearman_return_t *ret_ptr)
 
148
{
 
149
  size_t pack_size;
 
150
  gearman_protocol_http_st *http;
 
151
 
 
152
  http= (gearman_protocol_http_st *)gearman_con_protocol_data(con);
 
153
 
 
154
  if (packet->command != GEARMAN_COMMAND_WORK_COMPLETE &&
 
155
      packet->command != GEARMAN_COMMAND_WORK_FAIL &&
 
156
      (http->background == false ||
 
157
       packet->command != GEARMAN_COMMAND_JOB_CREATED))
 
158
  {
 
159
    *ret_ptr= GEARMAN_IGNORE_PACKET;
 
160
    return 0;
 
161
  }
 
162
 
 
163
  pack_size= (size_t)snprintf((char *)data, data_size,
 
164
                              "HTTP/1.0 200 OK\r\n"
 
165
                              "X-Gearman-Job-Handle: %.*s\r\n"
 
166
                              "Content-Length: %"PRIu64"\r\n"
 
167
                              "Server: Gearman/" PACKAGE_VERSION "\r\n"
 
168
                              "\r\n",
 
169
                              packet->command == GEARMAN_COMMAND_JOB_CREATED ?
 
170
                              (uint32_t)packet->arg_size[0] :
 
171
                              (uint32_t)packet->arg_size[0] - 1,
 
172
                              packet->arg[0],
 
173
                              (uint64_t)packet->data_size);
 
174
 
 
175
  if (pack_size > data_size)
 
176
  {
 
177
    *ret_ptr= GEARMAN_FLUSH_DATA;
 
178
    return 0;
 
179
  }
 
180
 
 
181
  if (!(http->keep_alive))
 
182
    gearman_con_set_options(con, GEARMAN_CON_CLOSE_AFTER_FLUSH, 1);
 
183
 
 
184
  *ret_ptr= GEARMAN_SUCCESS;
 
185
  return pack_size;
 
186
}
 
187
 
 
188
static size_t _http_unpack(gearman_packet_st *packet, gearman_con_st *con,
 
189
                           const void *data, size_t data_size,
 
190
                           gearman_return_t *ret_ptr)
 
191
{
 
192
  gearman_protocol_http_st *http;
 
193
  size_t offset= 0;
 
194
  const char *request;
 
195
  size_t request_size;
 
196
  const char *method;
 
197
  ptrdiff_t method_size;
 
198
  const char *uri;
 
199
  ptrdiff_t uri_size;
 
200
  const char *version;
 
201
  size_t version_size;
 
202
  const char *header;
 
203
  size_t header_size;
 
204
  char content_length[11]; /* 11 bytes to fit max display length of uint32_t */
 
205
  const char *unique= "-";
 
206
  size_t unique_size= 2;
 
207
  gearman_job_priority_t priority= GEARMAN_JOB_PRIORITY_NORMAL;
 
208
 
 
209
  /* Get the request line first. */
 
210
  request= _http_line(data, data_size, &request_size, &offset);
 
211
  if (request == NULL || request_size == 0)
 
212
  {
 
213
    *ret_ptr= GEARMAN_IO_WAIT;
 
214
    return offset;
 
215
  }
 
216
 
 
217
  http= (gearman_protocol_http_st *)gearman_con_protocol_data(con);
 
218
  http->background= false;
 
219
  http->keep_alive= false;
 
220
 
 
221
  /* Parse out the method, URI, and HTTP version from the request line. */
 
222
  method= request;
 
223
  uri= memchr(request, ' ', request_size);
 
224
  if (uri == NULL)
 
225
  {
 
226
    GEARMAN_ERROR_SET(packet->gearman, "_http_unpack", "bad request line: %.*s",
 
227
                      (uint32_t)request_size, request);
 
228
    *ret_ptr= GEARMAN_INVALID_PACKET;
 
229
    return 0;
 
230
  }
 
231
 
 
232
  method_size= uri - request;
 
233
  if ((method_size != 3 ||
 
234
       (strncasecmp(method, "GET", 3) && strncasecmp(method, "PUT", 3))) &&
 
235
      (method_size != 4 || strncasecmp(method, "POST", 4)))
 
236
  {
 
237
    GEARMAN_ERROR_SET(packet->gearman, "_http_unpack", "bad method: %.*s",
 
238
                      (uint32_t)method_size, method);
 
239
    *ret_ptr= GEARMAN_INVALID_PACKET;
 
240
    return 0;
 
241
  }
 
242
 
 
243
  while (*uri == ' ')
 
244
    uri++;
 
245
 
 
246
  while (*uri == '/')
 
247
    uri++;
 
248
 
 
249
  version= memchr(uri, ' ', request_size - (size_t)(uri - request));
 
250
  if (version == NULL)
 
251
  {
 
252
    GEARMAN_ERROR_SET(packet->gearman, "_http_unpack", "bad request line: %.*s",
 
253
                      (uint32_t)request_size, request);
 
254
    *ret_ptr= GEARMAN_INVALID_PACKET;
 
255
    return 0;
 
256
  }
 
257
 
 
258
  uri_size= version - uri;
 
259
  if (uri_size == 0)
 
260
  {
 
261
    GEARMAN_ERROR_SET(packet->gearman, "_http_unpack",
 
262
                      "must give function name in URI")
 
263
    *ret_ptr= GEARMAN_INVALID_PACKET;
 
264
    return 0;
 
265
  }
 
266
 
 
267
  while (*version == ' ')
 
268
    version++;
 
269
 
 
270
  version_size= request_size - (size_t)(version - request);
 
271
 
 
272
  if (version_size == 8 && !strncasecmp(version, "HTTP/1.1", 8))
 
273
    http->keep_alive= true;
 
274
  else if (version_size != 8 || strncasecmp(version, "HTTP/1.0", 8))
 
275
  {
 
276
    GEARMAN_ERROR_SET(packet->gearman, "_http_unpack", "bad version: %.*s",
 
277
                      (uint32_t)version_size, version);
 
278
    *ret_ptr= GEARMAN_INVALID_PACKET;
 
279
    return 0;
 
280
  }
 
281
 
 
282
  /* Loop through all the headers looking for ones of interest. */
 
283
  while ((header= _http_line(data, data_size, &header_size, &offset)) != NULL)
 
284
  {
 
285
    if (header_size == 0)
 
286
      break;
 
287
 
 
288
    if (header_size > 16 && !strncasecmp(header, "Content-Length: ", 16))
 
289
    {
 
290
      if ((method_size == 4 && !strncasecmp(method, "POST", 4)) ||
 
291
          (method_size == 3 && !strncasecmp(method, "PUT", 3)))
 
292
      {
 
293
        snprintf(content_length, 11, "%.*s", (uint32_t)header_size - 16,
 
294
                 header + 16);
 
295
        packet->data_size= (size_t)atoi(content_length);
 
296
      }
 
297
    }
 
298
    else if (header_size == 22 &&
 
299
             !strncasecmp(header, "Connection: Keep-Alive", 22))
 
300
    {
 
301
      http->keep_alive= true;
 
302
    }
 
303
    else if (header_size > 18 && !strncasecmp(header, "X-Gearman-Unique: ", 18))
 
304
    {
 
305
      unique= header + 18;
 
306
      unique_size= header_size - 18;
 
307
    }
 
308
    else if (header_size == 26 &&
 
309
             !strncasecmp(header, "X-Gearman-Background: true", 26))
 
310
    {
 
311
      http->background= true;
 
312
    }
 
313
    else if (header_size == 24 &&
 
314
             !strncasecmp(header, "X-Gearman-Priority: high", 24))
 
315
    {
 
316
      priority= GEARMAN_JOB_PRIORITY_HIGH;
 
317
    }
 
318
    else if (header_size == 23 &&
 
319
             !strncasecmp(header, "X-Gearman-Priority: low", 23))
 
320
    {
 
321
      priority= GEARMAN_JOB_PRIORITY_LOW;
 
322
    }
 
323
  }
 
324
 
 
325
  /* Make sure we received the end of headers. */
 
326
  if (header == NULL)
 
327
  {
 
328
    *ret_ptr= GEARMAN_IO_WAIT;
 
329
    return 0;
 
330
  }
 
331
 
 
332
  /* Request and all headers complete, build a packet based on HTTP request. */
 
333
  packet->magic= GEARMAN_MAGIC_REQUEST;
 
334
 
 
335
  if (http->background)
 
336
  {
 
337
    if (priority == GEARMAN_JOB_PRIORITY_NORMAL)
 
338
      packet->command= GEARMAN_COMMAND_SUBMIT_JOB_BG;
 
339
    if (priority == GEARMAN_JOB_PRIORITY_HIGH)
 
340
      packet->command= GEARMAN_COMMAND_SUBMIT_JOB_HIGH_BG;
 
341
    else
 
342
      packet->command= GEARMAN_COMMAND_SUBMIT_JOB_LOW_BG;
 
343
  }
 
344
  else
 
345
  {
 
346
    if (priority == GEARMAN_JOB_PRIORITY_NORMAL)
 
347
      packet->command= GEARMAN_COMMAND_SUBMIT_JOB;
 
348
    if (priority == GEARMAN_JOB_PRIORITY_HIGH)
 
349
      packet->command= GEARMAN_COMMAND_SUBMIT_JOB_HIGH;
 
350
    else
 
351
      packet->command= GEARMAN_COMMAND_SUBMIT_JOB_LOW;
 
352
  }
 
353
 
 
354
  *ret_ptr= gearman_packet_pack_header(packet);
 
355
  if (*ret_ptr != GEARMAN_SUCCESS)
 
356
    return 0;
 
357
 
 
358
  *ret_ptr= gearman_packet_add_arg(packet, uri, (size_t)uri_size + 1);
 
359
  if (*ret_ptr != GEARMAN_SUCCESS)
 
360
    return 0;
 
361
 
 
362
  *ret_ptr= gearman_packet_add_arg(packet, unique, unique_size + 1);
 
363
  if (*ret_ptr != GEARMAN_SUCCESS)
 
364
    return 0;
 
365
 
 
366
  /* Make sure function and unique are NULL terminated. */
 
367
  packet->arg[0][uri_size]= 0;
 
368
  packet->arg[1][unique_size]= 0;
 
369
 
 
370
  *ret_ptr= GEARMAN_SUCCESS;
 
371
  return offset;
 
372
}
 
373
 
 
374
static const char *_http_line(const void *data, size_t data_size,
 
375
                              size_t *line_size, size_t *offset)
 
376
{
 
377
  const char *start= (const char *)data + *offset;
 
378
  const char *end;
 
379
 
 
380
  end= memchr(start, '\n', data_size - *offset);
 
381
  if (end == NULL)
 
382
    return NULL;
 
383
 
 
384
  *offset+= (size_t)(end - start) + 1;
 
385
 
 
386
  if (end != start && *(end - 1) == '\r')
 
387
    end--;
 
388
 
 
389
  *line_size= (size_t)(end - start);
 
390
 
 
391
  return start;
 
392
}