« back to all changes in this revision
Viewing changes to outgoing.c
- browse files at revision 3.2.6
- compare with another revision
- download diff
- download tarball
- view history from revision 3.2.6
- Committer: Bazaar Package Importer
- Author(s): Peter Samuelson
- Date: 2011-06-03 03:18:07 UTC
- mfrom: (1.2.4 upstream)
- mto: (3.3.1 sid)
- mto: This revision was merged to the branch mainline in revision 11.
- Revision ID: james.westby@ubuntu.com-20110603031807-3vq82t5lzw692our
Tags: 0.7.2-1
* New upstream release.
- patches/no-export-vars: delete, now upstream.
* New ABI:
- patches/abi-0.x: New patch to change upstream SONAME.
- Rename libserf-0-0 to libserf0.7.
- Rename libserf-0-0-dev to libserf-dev while we're at it.
* Policy 3.9.1: one instance of s/Conflicts/Breaks/.
* "Upgrade" to source format 1.0.
* Add some Depends: ${misc:Depends}; thanks, Lintian.
- patches/no-export-vars: delete, now upstream.
* New ABI:
- patches/abi-0.x: New patch to change upstream SONAME.
- Rename libserf-0-0 to libserf0.7.
- Rename libserf-0-0-dev to libserf-dev while we're at it.
* Policy 3.9.1: one instance of s/Conflicts/Breaks/.
* "Upgrade" to source format 1.0.
* Add some Depends: ${misc:Depends}; thanks, Lintian.
- files added:
- auth
- debian/source
- files removed:
- debian/libserf-0-0-dev.docs
- files modified:
- CHANGES
added
removed
outgoing.c
1
/* Copyright 2002-2004 Justin Erenkrantz and Greg Stein
2
*
3
* Licensed under the Apache License, Version 2.0 (the "License");
4
* you may not use this file except in compliance with the License.
5
* You may obtain a copy of the License at
6
*
7
* http://www.apache.org/licenses/LICENSE-2.0
8
*
9
* Unless required by applicable law or agreed to in writing, software
10
* distributed under the License is distributed on an "AS IS" BASIS,
11
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
* See the License for the specific language governing permissions and
13
* limitations under the License.
14
*/
15
16
#include <apr_pools.h>
17
#include <apr_poll.h>
18
#include <apr_version.h>
19
20
#include "serf.h"
21
#include "serf_bucket_util.h"
22
23
#include "serf_private.h"
24
25
/* cleanup for sockets */
26
static apr_status_t clean_skt(void *data)
27
{
28
serf_connection_t *conn = data;
29
apr_status_t status = APR_SUCCESS;
30
31
if (conn->skt) {
32
status = apr_socket_close(conn->skt);
33
conn->skt = NULL;
34
}
35
36
return status;
37
}
38
39
static apr_status_t clean_resp(void *data)
40
{
41
serf_request_t *req = data;
42
43
/* This pool just got cleared/destroyed. Don't try to destroy the pool
44
* (again) when the request is canceled.
45
*/
46
req->respool = NULL;
47
48
return APR_SUCCESS;
49
}
50
51
/* cleanup for conns */
52
static apr_status_t clean_conn(void *data)
53
{
54
serf_connection_t *conn = data;
55
56
serf_connection_close(conn);
57
58
return APR_SUCCESS;
59
}
60
61
/* Update the pollset for this connection. We tweak the pollset based on
62
* whether we want to read and/or write, given conditions within the
63
* connection. If the connection is not (yet) in the pollset, then it
64
* will be added.
65
*/
66
apr_status_t serf__conn_update_pollset(serf_connection_t *conn)
67
{
68
serf_context_t *ctx = conn->ctx;
69
apr_status_t status;
70
apr_pollfd_t desc = { 0 };
71
72
if (!conn->skt) {
73
return APR_SUCCESS;
74
}
75
76
/* Remove the socket from the poll set. */
77
desc.desc_type = APR_POLL_SOCKET;
78
desc.desc.s = conn->skt;
79
desc.reqevents = conn->reqevents;
80
81
status = ctx->pollset_rm(ctx->pollset_baton,
82
&desc, conn);
83
if (status && !APR_STATUS_IS_NOTFOUND(status))
84
return status;
85
86
/* Now put it back in with the correct read/write values. */
87
desc.reqevents = APR_POLLHUP | APR_POLLERR;
88
if (conn->requests) {
89
/* If there are any outstanding events, then we want to read. */
90
/* ### not true. we only want to read IF we have sent some data */
91
desc.reqevents |= APR_POLLIN;
92
93
/* If the connection has unwritten data, or there are any requests
94
* that still have buckets to write out, then we want to write.
95
*/
96
if (conn->vec_len)
97
desc.reqevents |= APR_POLLOUT;
98
else {
99
serf_request_t *request = conn->requests;
100
101
if ((conn->probable_keepalive_limit &&
102
conn->completed_requests > conn->probable_keepalive_limit) ||
103
(conn->max_outstanding_requests &&
104
conn->completed_requests - conn->completed_responses >=
105
conn->max_outstanding_requests)) {
106
/* we wouldn't try to write any way right now. */
107
}
108
else {
109
while (request != NULL && request->req_bkt == NULL &&
110
request->written)
111
request = request->next;
112
if (request != NULL)
113
desc.reqevents |= APR_POLLOUT;
114
}
115
}
116
}
117
118
/* If we can have async responses, always look for something to read. */
119
if (conn->async_responses) {
120
desc.reqevents |= APR_POLLIN;
121
}
122
123
/* save our reqevents, so we can pass it in to remove later. */
124
conn->reqevents = desc.reqevents;
125
126
/* Note: even if we don't want to read/write this socket, we still
127
* want to poll it for hangups and errors.
128
*/
129
return ctx->pollset_add(ctx->pollset_baton,
130
&desc, &conn->baton);
131
}
132
133
#ifdef SERF_DEBUG_BUCKET_USE
134
135
/* Make sure all response buckets were drained. */
136
static void check_buckets_drained(serf_connection_t *conn)
137
{
138
serf_request_t *request = conn->requests;
139
140
for ( ; request ; request = request->next ) {
141
if (request->resp_bkt != NULL) {
142
/* ### crap. can't do this. this allocator may have un-drained
143
* ### REQUEST buckets.
144
*/
145
/* serf_debug__entered_loop(request->resp_bkt->allocator); */
146
/* ### for now, pretend we closed the conn (resets the tracking) */
147
serf_debug__closed_conn(request->resp_bkt->allocator);
148
}
149
}
150
}
151
152
#endif
153
154
/* Create and connect sockets for any connections which don't have them
155
* yet. This is the core of our lazy-connect behavior.
156
*/
157
apr_status_t serf__open_connections(serf_context_t *ctx)
158
{
159
int i;
160
161
for (i = ctx->conns->nelts; i--; ) {
162
serf_connection_t *conn = GET_CONN(ctx, i);
163
apr_status_t status;
164
apr_socket_t *skt;
165
apr_sockaddr_t *serv_addr;
166
167
conn->seen_in_pollset = 0;
168
169
if (conn->skt != NULL) {
170
#ifdef SERF_DEBUG_BUCKET_USE
171
check_buckets_drained(conn);
172
#endif
173
continue;
174
}
175
176
/* Delay opening until we have something to deliver! */
177
if (conn->requests == NULL) {
178
continue;
179
}
180
181
apr_pool_clear(conn->skt_pool);
182
apr_pool_cleanup_register(conn->skt_pool, conn, clean_skt, clean_skt);
183
184
/* Do we have to connect to a proxy server? */
185
if (ctx->proxy_address)
186
serv_addr = ctx->proxy_address;
187
else
188
serv_addr = conn->address;
189
190
if ((status = apr_socket_create(&skt, serv_addr->family,
191
SOCK_STREAM,
192
#if APR_MAJOR_VERSION > 0
193
APR_PROTO_TCP,
194
#endif
195
conn->skt_pool)) != APR_SUCCESS)
196
return status;
197
198
/* Set the socket to be non-blocking */
199
if ((status = apr_socket_timeout_set(skt, 0)) != APR_SUCCESS)
200
return status;
201
202
/* Disable Nagle's algorithm */
203
if ((status = apr_socket_opt_set(skt,
204
APR_TCP_NODELAY, 1)) != APR_SUCCESS)
205
return status;
206
207
/* Configured. Store it into the connection now. */
208
conn->skt = skt;
209
210
/* Now that the socket is set up, let's connect it. This should
211
* return immediately.
212
*/
213
if ((status = apr_socket_connect(skt,
214
serv_addr)) != APR_SUCCESS) {
215
if (!APR_STATUS_IS_EINPROGRESS(status))
216
return status;
217
}
218
219
/* Flag our pollset as dirty now that we have a new socket. */
220
conn->dirty_conn = 1;
221
ctx->dirty_pollset = 1;
222
223
/* If the authentication was already started on another connection,
224
prepare this connection (it might be possible to skip some
225
part of the handshaking). */
226
if (ctx->proxy_address) {
227
if (conn->ctx->proxy_authn_info.scheme)
228
conn->ctx->proxy_authn_info.scheme->init_conn_func(407, conn,
229
conn->pool);
230
}
231
232
if (conn->ctx->authn_info.scheme)
233
conn->ctx->authn_info.scheme->init_conn_func(401, conn,
234
conn->pool);
235
}
236
237
return APR_SUCCESS;
238
}
239
240
static apr_status_t no_more_writes(serf_connection_t *conn,
241
serf_request_t *request)
242
{
243
/* Note that we should hold new requests until we open our new socket. */
244
conn->closing = 1;
245
246
/* We can take the *next* request in our list and assume it hasn't
247
* been written yet and 'save' it for the new socket.
248
*/
249
conn->hold_requests = request->next;
250
conn->hold_requests_tail = conn->requests_tail;
251
request->next = NULL;
252
conn->requests_tail = request;
253
254
/* Clear our iovec. */
255
conn->vec_len = 0;
256
257
/* Update the pollset to know we don't want to write on this socket any
258
* more.
259
*/
260
conn->dirty_conn = 1;
261
conn->ctx->dirty_pollset = 1;
262
return APR_SUCCESS;
263
}
264
265
/* Read the 'Connection' header from the response. Return SERF_ERROR_CLOSING if
266
* the header contains value 'close' indicating the server is closing the
267
* connection right after this response.
268
* Otherwise returns APR_SUCCESS.
269
*/
270
static apr_status_t is_conn_closing(serf_bucket_t *response)
271
{
272
serf_bucket_t *hdrs;
273
const char *val;
274
275
hdrs = serf_bucket_response_get_headers(response);
276
val = serf_bucket_headers_get(hdrs, "Connection");
277
if (val && strcasecmp("close", val) == 0)
278
{
279
return SERF_ERROR_CLOSING;
280
}
281
282
return APR_SUCCESS;
283
}
284
285
static void link_requests(serf_request_t **list, serf_request_t **tail,
286
serf_request_t *request)
287
{
288
if (*list == NULL) {
289
*list = request;
290
*tail = request;
291
}
292
else {
293
(*tail)->next = request;
294
*tail = request;
295
}
296
}
297
298
static apr_status_t destroy_request(serf_request_t *request)
299
{
300
serf_connection_t *conn = request->conn;
301
302
/* The request and response buckets are no longer needed,
303
nor is the request's pool. */
304
if (request->resp_bkt) {
305
serf_debug__closed_conn(request->resp_bkt->allocator);
306
serf_bucket_destroy(request->resp_bkt);
307
}
308
if (request->req_bkt) {
309
serf_debug__closed_conn(request->req_bkt->allocator);
310
serf_bucket_destroy(request->req_bkt);
311
}
312
313
serf_debug__bucket_alloc_check(request->allocator);
314
if (request->respool) {
315
apr_pool_destroy(request->respool);
316
}
317
318
serf_bucket_mem_free(conn->allocator, request);
319
320
return APR_SUCCESS;
321
}
322
323
static apr_status_t cancel_request(serf_request_t *request,
324
serf_request_t **list,
325
int notify_request)
326
{
327
/* If we haven't run setup, then we won't have a handler to call. */
328
if (request->handler && notify_request) {
329
/* We actually don't care what the handler returns.
330
* We have bigger matters at hand.
331
*/
332
(*request->handler)(request, NULL, request->handler_baton,
333
request->respool);
334
}
335
336
if (*list == request) {
337
*list = request->next;
338
}
339
else {
340
serf_request_t *scan = *list;
341
342
while (scan->next && scan->next != request)
343
scan = scan->next;
344
345
if (scan->next) {
346
scan->next = scan->next->next;
347
}
348
}
349
350
return destroy_request(request);
351
}
352
353
static apr_status_t remove_connection(serf_context_t *ctx,
354
serf_connection_t *conn)
355
{
356
apr_pollfd_t desc = { 0 };
357
358
desc.desc_type = APR_POLL_SOCKET;
359
desc.desc.s = conn->skt;
360
desc.reqevents = conn->reqevents;
361
362
return ctx->pollset_rm(ctx->pollset_baton,
363
&desc, conn);
364
}
365
366
static void destroy_ostream(serf_connection_t *conn)
367
{
368
if (conn->ostream_head != NULL) {
369
serf_bucket_destroy(conn->ostream_head);
370
conn->ostream_head = NULL;
371
conn->ostream_tail = NULL;
372
}
373
}
374
375
/* A socket was closed, inform the application. */
376
static void handle_conn_closed(serf_connection_t *conn, apr_status_t status)
377
{
378
(*conn->closed)(conn, conn->closed_baton, status,
379
conn->pool);
380
}
381
382
static apr_status_t reset_connection(serf_connection_t *conn,
383
int requeue_requests)
384
{
385
serf_context_t *ctx = conn->ctx;
386
apr_status_t status;
387
serf_request_t *old_reqs, *held_reqs, *held_reqs_tail;
388
389
conn->probable_keepalive_limit = conn->completed_responses;
390
conn->completed_requests = 0;
391
conn->completed_responses = 0;
392
393
old_reqs = conn->requests;
394
held_reqs = conn->hold_requests;
395
held_reqs_tail = conn->hold_requests_tail;
396
397
if (conn->closing) {
398
conn->hold_requests = NULL;
399
conn->hold_requests_tail = NULL;
400
conn->closing = 0;
401
}
402
403
conn->requests = NULL;
404
conn->requests_tail = NULL;
405
406
while (old_reqs) {
407
/* If we haven't started to write the connection, bring it over
408
* unchanged to our new socket. Otherwise, call the cancel function.
409
*/
410
if (requeue_requests && !old_reqs->written) {
411
serf_request_t *req = old_reqs;
412
old_reqs = old_reqs->next;
413
req->next = NULL;
414
link_requests(&conn->requests, &conn->requests_tail, req);
415
}
416
else {
417
cancel_request(old_reqs, &old_reqs, requeue_requests);
418
}
419
}
420
421
if (conn->requests_tail) {
422
conn->requests_tail->next = held_reqs;
423
}
424
else {
425
conn->requests = held_reqs;
426
}
427
if (held_reqs_tail) {
428
conn->requests_tail = held_reqs_tail;
429
}
430
431
if (conn->skt != NULL) {
432
remove_connection(ctx, conn);
433
status = apr_socket_close(conn->skt);
434
if (conn->closed != NULL) {
435
handle_conn_closed(conn, status);
436
}
437
conn->skt = NULL;
438
}
439
440
if (conn->stream != NULL) {
441
serf_bucket_destroy(conn->stream);
442
conn->stream = NULL;
443
}
444
445
destroy_ostream(conn);
446
447
/* Don't try to resume any writes */
448
conn->vec_len = 0;
449
450
conn->dirty_conn = 1;
451
conn->ctx->dirty_pollset = 1;
452
453
conn->status = APR_SUCCESS;
454
455
/* Let our context know that we've 'reset' the socket already. */
456
conn->seen_in_pollset |= APR_POLLHUP;
457
458
/* Found the connection. Closed it. All done. */
459
return APR_SUCCESS;
460
}
461
462
static apr_status_t socket_writev(serf_connection_t *conn)
463
{
464
apr_size_t written;
465
apr_status_t status;
466
467
status = apr_socket_sendv(conn->skt, conn->vec,
468
conn->vec_len, &written);
469
470
/* did we write everything? */
471
if (written) {
472
apr_size_t len = 0;
473
int i;
474
475
for (i = 0; i < conn->vec_len; i++) {
476
len += conn->vec[i].iov_len;
477
if (written < len) {
478
if (i) {
479
memmove(conn->vec, &conn->vec[i],
480
sizeof(struct iovec) * (conn->vec_len - i));
481
conn->vec_len -= i;
482
}
483
conn->vec[0].iov_base = (char *)conn->vec[0].iov_base + (conn->vec[0].iov_len - (len - written));
484
conn->vec[0].iov_len = len - written;
485
break;
486
}
487
}
488
if (len == written) {
489
conn->vec_len = 0;
490
}
491
492
/* Log progress information */
493
serf__context_progress_delta(conn->ctx, 0, written);
494
}
495
496
return status;
497
}
498
499
static apr_status_t detect_eof(void *baton, serf_bucket_t *aggregate_bucket)
500
{
501
serf_connection_t *conn = baton;
502
conn->hit_eof = 1;
503
return APR_EAGAIN;
504
}
505
506
static apr_status_t do_conn_setup(serf_connection_t *conn)
507
{
508
apr_status_t status;
509
serf_bucket_t *ostream;
510
511
if (conn->ostream_head == NULL) {
512
conn->ostream_head = serf_bucket_aggregate_create(conn->allocator);
513
}
514
515
if (conn->ostream_tail == NULL) {
516
conn->ostream_tail = serf__bucket_stream_create(conn->allocator,
517
detect_eof,
518
conn);
519
}
520
521
ostream = conn->ostream_tail;
522
523
status = (*conn->setup)(conn->skt,
524
&conn->stream,
525
&ostream,
526
conn->setup_baton,
527
conn->pool);
528
if (status) {
529
/* extra destroy here since it wasn't added to the head bucket yet. */
530
serf_bucket_destroy(conn->ostream_tail);
531
destroy_ostream(conn);
532
return status;
533
}
534
535
serf_bucket_aggregate_append(conn->ostream_head,
536
ostream);
537
538
return status;
539
}
540
541
/* write data out to the connection */
542
static apr_status_t write_to_connection(serf_connection_t *conn)
543
{
544
serf_request_t *request = conn->requests;
545
546
if (conn->probable_keepalive_limit &&
547
conn->completed_requests > conn->probable_keepalive_limit) {
548
/* backoff for now. */
549
return APR_SUCCESS;
550
}
551
552
/* Find a request that has data which needs to be delivered. */
553
while (request != NULL &&
554
request->req_bkt == NULL && request->written)
555
request = request->next;
556
557
/* assert: request != NULL || conn->vec_len */
558
559
/* Keep reading and sending until we run out of stuff to read, or
560
* writing would block.
561
*/
562
while (1) {
563
int stop_reading = 0;
564
apr_status_t status;
565
apr_status_t read_status;
566
567
if (conn->max_outstanding_requests &&
568
conn->completed_requests -
569
conn->completed_responses >= conn->max_outstanding_requests) {
570
/* backoff for now. */
571
return APR_SUCCESS;
572
}
573
574
/* If we have unwritten data, then write what we can. */
575
while (conn->vec_len) {
576
status = socket_writev(conn);
577
578
/* If the write would have blocked, then we're done. Don't try
579
* to write anything else to the socket.
580
*/
581
if (APR_STATUS_IS_EAGAIN(status))
582
return APR_SUCCESS;
583
if (APR_STATUS_IS_EPIPE(status))
584
return no_more_writes(conn, request);
585
if (status)
586
return status;
587
}
588
/* ### can we have a short write, yet no EAGAIN? a short write
589
### would imply unwritten_len > 0 ... */
590
/* assert: unwritten_len == 0. */
591
592
/* We may need to move forward to a request which has something
593
* to write.
594
*/
595
while (request != NULL &&
596
request->req_bkt == NULL && request->written)
597
request = request->next;
598
599
if (request == NULL) {
600
/* No more requests (with data) are registered with the
601
* connection. Let's update the pollset so that we don't
602
* try to write to this socket again.
603
*/
604
conn->dirty_conn = 1;
605
conn->ctx->dirty_pollset = 1;
606
return APR_SUCCESS;
607
}
608
609
/* If the connection does not have an associated bucket, then
610
* call the setup callback to get one.
611
*/
612
if (conn->stream == NULL) {
613
status = do_conn_setup(conn);
614
if (status) {
615
return status;
616
}
617
}
618
619
if (request->req_bkt == NULL) {
620
/* Now that we are about to serve the request, allocate a pool. */
621
apr_pool_create(&request->respool, conn->pool);
622
request->allocator = serf_bucket_allocator_create(request->respool,
623
NULL, NULL);
624
apr_pool_cleanup_register(request->respool, request,
625
clean_resp, clean_resp);
626
627
/* Fill in the rest of the values for the request. */
628
read_status = request->setup(request, request->setup_baton,
629
&request->req_bkt,
630
&request->acceptor,
631
&request->acceptor_baton,
632
&request->handler,
633
&request->handler_baton,
634
request->respool);
635
636
if (read_status) {
637
/* Something bad happened. Propagate any errors. */
638
return read_status;
639
}
640
641
request->written = 1;
642
serf_bucket_aggregate_append(conn->ostream_tail, request->req_bkt);
643
}
644
645
/* ### optimize at some point by using read_for_sendfile */
646
read_status = serf_bucket_read_iovec(conn->ostream_head,
647
SERF_READ_ALL_AVAIL,
648
IOV_MAX,
649
conn->vec,
650
&conn->vec_len);
651
652
if (!conn->hit_eof) {
653
if (APR_STATUS_IS_EAGAIN(read_status)) {
654
/* We read some stuff, but should not try to read again. */
655
stop_reading = 1;
656
657
/* ### we should avoid looking for writability for a while so
658
### that (hopefully) something will appear in the bucket so
659
### we can actually write something. otherwise, we could
660
### end up in a CPU spin: socket wants something, but we
661
### don't have anything (and keep returning EAGAIN)
662
*/
663
}
664
else if (read_status && !APR_STATUS_IS_EOF(read_status)) {
665
/* Something bad happened. Propagate any errors. */
666
return read_status;
667
}
668
}
669
670
/* If we got some data, then deliver it. */
671
/* ### what to do if we got no data?? is that a problem? */
672
if (conn->vec_len > 0) {
673
status = socket_writev(conn);
674
675
/* If we can't write any more, or an error occurred, then
676
* we're done here.
677
*/
678
if (APR_STATUS_IS_EAGAIN(status))
679
return APR_SUCCESS;
680
if (APR_STATUS_IS_EPIPE(status))
681
return no_more_writes(conn, request);
682
if (APR_STATUS_IS_ECONNRESET(status)) {
683
return no_more_writes(conn, request);
684
}
685
if (status)
686
return status;
687
}
688
689
if (conn->hit_eof &&
690
conn->vec_len == 0) {
691
/* If we hit the end of the request bucket and all of its data has
692
* been written, then clear it out to signify that we're done
693
* sending the request. On the next iteration through this loop:
694
* - if there are remaining bytes they will be written, and as the
695
* request bucket will be completely read it will be destroyed then.
696
* - we'll see if there are other requests that need to be sent
697
* ("pipelining").
698
*/
699
conn->hit_eof = 0;
700
request->req_bkt = NULL;
701
702
/* If our connection has async responses enabled, we're not
703
* going to get a reply back, so kill the request.
704
*/
705
if (conn->async_responses) {
706
conn->requests = request->next;
707
destroy_request(request);
708
}
709
710
conn->completed_requests++;
711
712
if (conn->probable_keepalive_limit &&
713
conn->completed_requests > conn->probable_keepalive_limit) {
714
/* backoff for now. */
715
stop_reading = 1;
716
}
717
}
718
719
if (stop_reading) {
720
return APR_SUCCESS;
721
}
722
}
723
/* NOTREACHED */
724
}
725
726
/* A response message was received from the server, so call
727
the handler as specified on the original request. */
728
static apr_status_t handle_response(serf_request_t *request,
729
apr_pool_t *pool)
730
{
731
apr_status_t status;
732
int consumed_response = 0;
733
734
/* Only enable the new authentication framework if the program has
735
* registered an authentication credential callback.
736
*
737
* This permits older Serf apps to still handle authentication
738
* themselves by not registering credential callbacks.
739
*/
740
if (request->conn->ctx->cred_cb) {
741
status = serf__handle_auth_response(&consumed_response,
742
request,
743
request->resp_bkt,
744
request->handler_baton,
745
pool);
746
747
/* If there was an error reading the response (maybe there wasn't
748
enough data available), don't bother passing the response to the
749
application.
750
751
If the authentication was tried, but failed, pass the response
752
to the application, maybe it can do better. */
753
if (APR_STATUS_IS_EOF(status) ||
754
APR_STATUS_IS_EAGAIN(status)) {
755
return status;
756
}
757
}
758
759
if (!consumed_response) {
760
return (*request->handler)(request,
761
request->resp_bkt,
762
request->handler_baton,
763
pool);
764
}
765
766
return status;
767
}
768
769
/* An async response message was received from the server. */
770
static apr_status_t handle_async_response(serf_connection_t *conn,
771
apr_pool_t *pool)
772
{
773
apr_status_t status;
774
775
if (conn->current_async_response == NULL) {
776
conn->current_async_response =
777
(*conn->async_acceptor)(NULL, conn->stream,
778
conn->async_acceptor_baton, pool);
779
}
780
781
status = (*conn->async_handler)(NULL, conn->current_async_response,
782
conn->async_handler_baton, pool);
783
784
if (APR_STATUS_IS_EOF(status)) {
785
serf_bucket_destroy(conn->current_async_response);
786
conn->current_async_response = NULL;
787
status = APR_SUCCESS;
788
}
789
790
return status;
791
}
792
793
/* read data from the connection */
794
static apr_status_t read_from_connection(serf_connection_t *conn)
795
{
796
apr_status_t status;
797
apr_pool_t *tmppool;
798
int close_connection = FALSE;
799
800
/* Whatever is coming in on the socket corresponds to the first request
801
* on our chain.
802
*/
803
serf_request_t *request = conn->requests;
804
805
/* assert: request != NULL */
806
807
if ((status = apr_pool_create(&tmppool, conn->pool)) != APR_SUCCESS)
808
goto error;
809
810
/* Invoke response handlers until we have no more work. */
811
while (1) {
812
apr_pool_clear(tmppool);
813
814
/* If the connection does not have an associated bucket, then
815
* call the setup callback to get one.
816
*/
817
if (conn->stream == NULL) {
818
status = do_conn_setup(conn);
819
if (status) {
820
goto error;
821
}
822
}
823
824
/* We have a different codepath when we can have async responses. */
825
if (conn->async_responses) {
826
/* TODO What about socket errors? */
827
status = handle_async_response(conn, tmppool);
828
if (APR_STATUS_IS_EAGAIN(status)) {
829
status = APR_SUCCESS;
830
goto error;
831
}
832
if (status) {
833
goto error;
834
}
835
continue;
836
}
837
838
/* We are reading a response for a request we haven't
839
* written yet!
840
*
841
* This shouldn't normally happen EXCEPT:
842
*
843
* 1) when the other end has closed the socket and we're
844
* pending an EOF return.
845
* 2) Doing the initial SSL handshake - we'll get EAGAIN
846
* as the SSL buckets will hide the handshake from us
847
* but not return any data.
848
*
849
* In these cases, we should not receive any actual user data.
850
*
851
* If we see an EOF (due to an expired timeout), we'll reset the
852
* connection and open a new one.
853
*/
854
if (request->req_bkt || !request->written) {
855
const char *data;
856
apr_size_t len;
857
858
status = serf_bucket_read(conn->stream, SERF_READ_ALL_AVAIL,
859
&data, &len);
860
861
if (!status && len) {
862
status = APR_EGENERAL;
863
}
864
else if (APR_STATUS_IS_EOF(status)) {
865
reset_connection(conn, 1);
866
status = APR_SUCCESS;
867
}
868
else if (APR_STATUS_IS_EAGAIN(status)) {
869
status = APR_SUCCESS;
870
}
871
872
goto error;
873
}
874
875
/* If the request doesn't have a response bucket, then call the
876
* acceptor to get one created.
877
*/
878
if (request->resp_bkt == NULL) {
879
request->resp_bkt = (*request->acceptor)(request, conn->stream,
880
request->acceptor_baton,
881
tmppool);
882
apr_pool_clear(tmppool);
883
}
884
885
status = handle_response(request, tmppool);
886
887
/* Some systems will not generate a HUP poll event so we have to
888
* handle the ECONNRESET issue here.
889
*/
890
if (APR_STATUS_IS_ECONNRESET(status) ||
891
status == SERF_ERROR_REQUEST_LOST) {
892
reset_connection(conn, 1);
893
status = APR_SUCCESS;
894
goto error;
895
}
896
897
/* If our response handler says it can't do anything more, we now
898
* treat that as a success.
899
*/
900
if (APR_STATUS_IS_EAGAIN(status)) {
901
status = APR_SUCCESS;
902
goto error;
903
}
904
905
/* If we received APR_SUCCESS, run this loop again. */
906
if (!status) {
907
continue;
908
}
909
910
close_connection = is_conn_closing(request->resp_bkt);
911
912
if (!APR_STATUS_IS_EOF(status) &&
913
close_connection != SERF_ERROR_CLOSING) {
914
/* Whether success, or an error, there is no more to do unless
915
* this request has been completed.
916
*/
917
goto error;
918
}
919
920
/* The request has been fully-delivered, and the response has
921
* been fully-read. Remove it from our queue and loop to read
922
* another response.
923
*/
924
conn->requests = request->next;
925
926
destroy_request(request);
927
928
request = conn->requests;
929
930
/* If we're truly empty, update our tail. */
931
if (request == NULL) {
932
conn->requests_tail = NULL;
933
}
934
935
conn->completed_responses++;
936
937
/* This means that we're being advised that the connection is done. */
938
if (close_connection == SERF_ERROR_CLOSING) {
939
reset_connection(conn, 1);
940
if (APR_STATUS_IS_EOF(status))
941
status = APR_SUCCESS;
942
goto error;
943
}
944
945
/* The server is suddenly deciding to serve more responses than we've
946
* seen before.
947
*
948
* Let our requests go.
949
*/
950
if (conn->probable_keepalive_limit &&
951
conn->completed_responses > conn->probable_keepalive_limit) {
952
conn->probable_keepalive_limit = 0;
953
}
954
955
/* If we just ran out of requests or have unwritten requests, then
956
* update the pollset. We don't want to read from this socket any
957
* more. We are definitely done with this loop, too.
958
*/
959
if (request == NULL || !request->written) {
960
conn->dirty_conn = 1;
961
conn->ctx->dirty_pollset = 1;
962
status = APR_SUCCESS;
963
goto error;
964
}
965
}
966
967
error:
968
apr_pool_destroy(tmppool);
969
return status;
970
}
971
972
/* process all events on the connection */
973
apr_status_t serf__process_connection(serf_connection_t *conn,
974
apr_int16_t events)
975
{
976
apr_status_t status;
977
978
/* POLLHUP/ERR should come after POLLIN so if there's an error message or
979
* the like sitting on the connection, we give the app a chance to read
980
* it before we trigger a reset condition.
981
*/
982
if ((events & APR_POLLIN) != 0) {
983
if ((status = read_from_connection(conn)) != APR_SUCCESS)
984
return status;
985
986
/* If we decided to reset our connection, return now as we don't
987
* want to write.
988
*/
989
if ((conn->seen_in_pollset & APR_POLLHUP) != 0) {
990
return APR_SUCCESS;
991
}
992
}
993
if ((events & APR_POLLHUP) != 0) {
994
return APR_ECONNRESET;
995
}
996
if ((events & APR_POLLERR) != 0) {
997
/* We might be talking to a buggy HTTP server that doesn't
998
* do lingering-close. (httpd < 2.1.8 does this.)
999
*
1000
* See:
1001
*
1002
* http://issues.apache.org/bugzilla/show_bug.cgi?id=35292
1003
*/
1004
if (!conn->probable_keepalive_limit) {
1005
return reset_connection(conn, 1);
1006
}
1007
return APR_EGENERAL;
1008
}
1009
if ((events & APR_POLLOUT) != 0) {
1010
if ((status = write_to_connection(conn)) != APR_SUCCESS)
1011
return status;
1012
}
1013
return APR_SUCCESS;
1014
}
1015
1016
serf_connection_t *serf_connection_create(
1017
serf_context_t *ctx,
1018
apr_sockaddr_t *address,
1019
serf_connection_setup_t setup,
1020
void *setup_baton,
1021
serf_connection_closed_t closed,
1022
void *closed_baton,
1023
apr_pool_t *pool)
1024
{
1025
serf_connection_t *conn = apr_pcalloc(pool, sizeof(*conn));
1026
1027
conn->ctx = ctx;
1028
conn->status = APR_SUCCESS;
1029
conn->address = address;
1030
conn->setup = setup;
1031
conn->setup_baton = setup_baton;
1032
conn->closed = closed;
1033
conn->closed_baton = closed_baton;
1034
conn->pool = pool;
1035
conn->allocator = serf_bucket_allocator_create(pool, NULL, NULL);
1036
conn->stream = NULL;
1037
conn->ostream_head = NULL;
1038
conn->ostream_tail = NULL;
1039
conn->baton.type = SERF_IO_CONN;
1040
conn->baton.u.conn = conn;
1041
conn->hit_eof = 0;
1042
1043
/* Create a subpool for our connection. */
1044
apr_pool_create(&conn->skt_pool, conn->pool);
1045
1046
/* register a cleanup */
1047
apr_pool_cleanup_register(conn->pool, conn, clean_conn, apr_pool_cleanup_null);
1048
1049
/* Add the connection to the context. */
1050
*(serf_connection_t **)apr_array_push(ctx->conns) = conn;
1051
1052
return conn;
1053
}
1054
1055
apr_status_t serf_connection_create2(
1056
serf_connection_t **conn,
1057
serf_context_t *ctx,
1058
apr_uri_t host_info,
1059
serf_connection_setup_t setup,
1060
void *setup_baton,
1061
serf_connection_closed_t closed,
1062
void *closed_baton,
1063
apr_pool_t *pool)
1064
{
1065
apr_status_t status;
1066
serf_connection_t *c;
1067
apr_sockaddr_t *host_address;
1068
1069
/* Support for HTTPS proxies is not implemented yet. */
1070
if (ctx->proxy_address && strcmp(host_info.scheme, "https") == 0)
1071
return APR_ENOTIMPL;
1072
1073
/* Parse the url, store the address of the server. */
1074
status = apr_sockaddr_info_get(&host_address,
1075
host_info.hostname,
1076
APR_UNSPEC, host_info.port, 0, pool);
1077
if (status)
1078
return status;
1079
1080
c = serf_connection_create(ctx, host_address, setup, setup_baton,
1081
closed, closed_baton, pool);
1082
1083
/* We're not interested in the path following the hostname. */
1084
c->host_url = apr_uri_unparse(c->pool,
1085
&host_info,
1086
APR_URI_UNP_OMITPATHINFO);
1087
c->host_info = host_info;
1088
1089
*conn = c;
1090
1091
return status;
1092
}
1093
1094
apr_status_t serf_connection_reset(
1095
serf_connection_t *conn)
1096
{
1097
return reset_connection(conn, 0);
1098
}
1099
1100
1101
apr_status_t serf_connection_close(
1102
serf_connection_t *conn)
1103
{
1104
int i;
1105
serf_context_t *ctx = conn->ctx;
1106
apr_status_t status;
1107
1108
for (i = ctx->conns->nelts; i--; ) {
1109
serf_connection_t *conn_seq = GET_CONN(ctx, i);
1110
1111
if (conn_seq == conn) {
1112
while (conn->requests) {
1113
serf_request_cancel(conn->requests);
1114
}
1115
if (conn->skt != NULL) {
1116
remove_connection(ctx, conn);
1117
status = apr_socket_close(conn->skt);
1118
if (conn->closed != NULL) {
1119
handle_conn_closed(conn, status);
1120
}
1121
conn->skt = NULL;
1122
}
1123
if (conn->stream != NULL) {
1124
serf_bucket_destroy(conn->stream);
1125
conn->stream = NULL;
1126
}
1127
1128
/* Remove the connection from the context. We don't want to
1129
* deal with it any more.
1130
*/
1131
if (i < ctx->conns->nelts - 1) {
1132
/* move later connections over this one. */
1133
memmove(
1134
&GET_CONN(ctx, i),
1135
&GET_CONN(ctx, i + 1),
1136
(ctx->conns->nelts - i - 1) * sizeof(serf_connection_t *));
1137
}
1138
--ctx->conns->nelts;
1139
1140
/* Found the connection. Closed it. All done. */
1141
return APR_SUCCESS;
1142
}
1143
}
1144
1145
/* We didn't find the specified connection. */
1146
/* ### doc talks about this w.r.t poll structures. use something else? */
1147
return APR_NOTFOUND;
1148
}
1149
1150
1151
void serf_connection_set_max_outstanding_requests(
1152
serf_connection_t *conn,
1153
unsigned int max_requests)
1154
{
1155
conn->max_outstanding_requests = max_requests;
1156
}
1157
1158
1159
void serf_connection_set_async_responses(
1160
serf_connection_t *conn,
1161
serf_response_acceptor_t acceptor,
1162
void *acceptor_baton,
1163
serf_response_handler_t handler,
1164
void *handler_baton)
1165
{
1166
conn->async_responses = 1;
1167
conn->async_acceptor = acceptor;
1168
conn->async_acceptor_baton = acceptor_baton;
1169
conn->async_handler = handler;
1170
conn->async_handler_baton = handler_baton;
1171
}
1172
1173
1174
serf_request_t *serf_connection_request_create(
1175
serf_connection_t *conn,
1176
serf_request_setup_t setup,
1177
void *setup_baton)
1178
{
1179
serf_request_t *request;
1180
1181
request = serf_bucket_mem_alloc(conn->allocator, sizeof(*request));
1182
request->conn = conn;
1183
request->setup = setup;
1184
request->setup_baton = setup_baton;
1185
request->handler = NULL;
1186
request->respool = NULL;
1187
request->req_bkt = NULL;
1188
request->resp_bkt = NULL;
1189
request->written = 0;
1190
request->next = NULL;
1191
1192
/* Link the request to the end of the request chain. */
1193
if (conn->closing) {
1194
link_requests(&conn->hold_requests, &conn->hold_requests_tail, request);
1195
}
1196
else {
1197
link_requests(&conn->requests, &conn->requests_tail, request);
1198
1199
/* Ensure our pollset becomes writable in context run */
1200
conn->ctx->dirty_pollset = 1;
1201
conn->dirty_conn = 1;
1202
}
1203
1204
return request;
1205
}
1206
1207
1208
serf_request_t *serf_connection_priority_request_create(
1209
serf_connection_t *conn,
1210
serf_request_setup_t setup,
1211
void *setup_baton)
1212
{
1213
serf_request_t *request;
1214
serf_request_t *iter, *prev;
1215
1216
request = serf_bucket_mem_alloc(conn->allocator, sizeof(*request));
1217
request->conn = conn;
1218
request->setup = setup;
1219
request->setup_baton = setup_baton;
1220
request->handler = NULL;
1221
request->respool = NULL;
1222
request->req_bkt = NULL;
1223
request->resp_bkt = NULL;
1224
request->written = 0;
1225
request->next = NULL;
1226
1227
/* Link the new request after the last written request, but before all
1228
upcoming requests. */
1229
if (conn->closing) {
1230
iter = conn->hold_requests;
1231
}
1232
else {
1233
iter = conn->requests;
1234
}
1235
prev = NULL;
1236
1237
/* Find a request that has data which needs to be delivered. */
1238
while (iter != NULL && iter->req_bkt == NULL && iter->written) {
1239
prev = iter;
1240
iter = iter->next;
1241
}
1242
1243
if (prev) {
1244
request->next = iter;
1245
prev->next = request;
1246
} else {
1247
request->next = iter;
1248
if (conn->closing) {
1249
conn->hold_requests = request;
1250
}
1251
else {
1252
conn->requests = request;
1253
}
1254
}
1255
1256
if (! conn->closing) {
1257
/* Ensure our pollset becomes writable in context run */
1258
conn->ctx->dirty_pollset = 1;
1259
conn->dirty_conn = 1;
1260
}
1261
1262
return request;
1263
}
1264
1265
1266
apr_status_t serf_request_cancel(serf_request_t *request)
1267
{
1268
return cancel_request(request, &request->conn->requests, 0);
1269
}
1270
1271
1272
apr_pool_t *serf_request_get_pool(const serf_request_t *request)
1273
{
1274
return request->respool;
1275
}
1276
1277
1278
serf_bucket_alloc_t *serf_request_get_alloc(
1279
const serf_request_t *request)
1280
{
1281
return request->allocator;
1282
}
1283
1284
1285
serf_connection_t *serf_request_get_conn(
1286
const serf_request_t *request)
1287
{
1288
return request->conn;
1289
}
1290
1291
1292
void serf_request_set_handler(
1293
serf_request_t *request,
1294
const serf_response_handler_t handler,
1295
const void **handler_baton)
1296
{
1297
request->handler = handler;
1298
request->handler_baton = handler_baton;
1299
}
1300
1301
1302
serf_bucket_t *serf_request_bucket_request_create(
1303
serf_request_t *request,
1304
const char *method,
1305
const char *uri,
1306
serf_bucket_t *body,
1307
serf_bucket_alloc_t *allocator)
1308
{
1309
serf_bucket_t *req_bkt, *hdrs_bkt;
1310
serf_connection_t *conn = request->conn;
1311
serf_context_t *ctx = conn->ctx;
1312
1313
req_bkt = serf_bucket_request_create(method, uri, body, allocator);
1314
hdrs_bkt = serf_bucket_request_get_headers(req_bkt);
1315
1316
/* Proxy? */
1317
if (ctx->proxy_address && conn->host_url)
1318
serf_bucket_request_set_root(req_bkt, conn->host_url);
1319
1320
if (conn->host_info.hostinfo)
1321
serf_bucket_headers_setn(hdrs_bkt, "Host",
1322
conn->host_info.hostinfo);
1323
1324
/* Setup server authorization headers */
1325
if (ctx->authn_info.scheme)
1326
ctx->authn_info.scheme->setup_request_func(401, conn, method, uri,
1327
hdrs_bkt);
1328
1329
/* Setup proxy authorization headers */
1330
if (ctx->proxy_authn_info.scheme)
1331
ctx->proxy_authn_info.scheme->setup_request_func(407, conn, method,
1332
uri, hdrs_bkt);
1333
1334
return req_bkt;
1335
}
Loggerhead is a web-based interface for Breezy
Version: 2.0.1