1 /*
2 * request_body.c : svn_ra_serf__request_body_t implementation
3 *
4 * ====================================================================
5 * Licensed to the Apache Software Foundation (ASF) under one
6 * or more contributor license agreements. See the NOTICE file
7 * distributed with this work for additional information
8 * regarding copyright ownership. The ASF licenses this file
9 * to you under the Apache License, Version 2.0 (the
10 * "License"); you may not use this file except in compliance
11 * with the License. You may obtain a copy of the License at
12 *
13 * http://www.apache.org/licenses/LICENSE-2.0
14 *
15 * Unless required by applicable law or agreed to in writing,
16 * software distributed under the License is distributed on an
17 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
18 * KIND, either express or implied. See the License for the
19 * specific language governing permissions and limitations
20 * under the License.
21 * ====================================================================
22 */
23
24 #include <serf.h>
25
26 #include "ra_serf.h"
27
28 struct svn_ra_serf__request_body_t
29 {
30 svn_stream_t *stream;
31 apr_size_t in_memory_size;
32 apr_size_t total_bytes;
33 serf_bucket_alloc_t *alloc;
34 serf_bucket_t *collect_bucket;
35 const void *all_data;
36 apr_file_t *file;
37 apr_pool_t *result_pool;
38 apr_pool_t *scratch_pool;
39 };
40
41 /* Fold all previously collected data in a single buffer allocated in
42 RESULT_POOL and clear all intermediate state. */
43 static const char *
allocate_all(svn_ra_serf__request_body_t * body,apr_pool_t * result_pool)44 allocate_all(svn_ra_serf__request_body_t *body,
45 apr_pool_t *result_pool)
46 {
47 char *buffer = apr_pcalloc(result_pool, body->total_bytes);
48 const char *data;
49 apr_size_t sz;
50 apr_status_t s;
51 apr_size_t remaining = body->total_bytes;
52 char *next = buffer;
53
54 while (!(s = serf_bucket_read(body->collect_bucket, remaining, &data, &sz)))
55 {
56 memcpy(next, data, sz);
57 remaining -= sz;
58 next += sz;
59
60 if (! remaining)
61 break;
62 }
63
64 if (!SERF_BUCKET_READ_ERROR(s))
65 {
66 memcpy(next, data, sz);
67 }
68
69 serf_bucket_destroy(body->collect_bucket);
70 body->collect_bucket = NULL;
71
72 return (s != APR_EOF) ? NULL : buffer;
73 }
74
75 /* Noop function. Make serf take care of freeing in error situations. */
serf_free_no_error(void * unfreed_baton,void * block)76 static void serf_free_no_error(void *unfreed_baton, void *block) {}
77
78 /* Stream write function for body creation. */
79 static svn_error_t *
request_body_stream_write(void * baton,const char * data,apr_size_t * len)80 request_body_stream_write(void *baton,
81 const char *data,
82 apr_size_t *len)
83 {
84 svn_ra_serf__request_body_t *b = baton;
85
86 if (!b->scratch_pool)
87 b->scratch_pool = svn_pool_create(b->result_pool);
88
89 if (b->file)
90 {
91 SVN_ERR(svn_io_file_write_full(b->file, data, *len, NULL,
92 b->scratch_pool));
93 svn_pool_clear(b->scratch_pool);
94
95 b->total_bytes += *len;
96 }
97 else if (*len + b->total_bytes > b->in_memory_size)
98 {
99 SVN_ERR(svn_io_open_unique_file3(&b->file, NULL, NULL,
100 svn_io_file_del_on_pool_cleanup,
101 b->result_pool, b->scratch_pool));
102
103 if (b->total_bytes)
104 {
105 const char *all = allocate_all(b, b->scratch_pool);
106
107 SVN_ERR(svn_io_file_write_full(b->file, all, b->total_bytes,
108 NULL, b->scratch_pool));
109 }
110
111 SVN_ERR(svn_io_file_write_full(b->file, data, *len, NULL,
112 b->scratch_pool));
113 b->total_bytes += *len;
114 }
115 else
116 {
117 if (!b->alloc)
118 b->alloc = serf_bucket_allocator_create(b->scratch_pool,
119 serf_free_no_error, NULL);
120
121 if (!b->collect_bucket)
122 b->collect_bucket = serf_bucket_aggregate_create(b->alloc);
123
124 serf_bucket_aggregate_append(b->collect_bucket,
125 serf_bucket_simple_copy_create(data, *len,
126 b->alloc));
127
128 b->total_bytes += *len;
129 }
130
131 return SVN_NO_ERROR;
132 }
133
134 /* Stream close function for collecting body. */
135 static svn_error_t *
request_body_stream_close(void * baton)136 request_body_stream_close(void *baton)
137 {
138 svn_ra_serf__request_body_t *b = baton;
139
140 if (b->file)
141 {
142 /* We need to flush the file, make it unbuffered (so that it can be
143 * zero-copied via mmap), and reset the position before attempting
144 * to deliver the file.
145 *
146 * N.B. If we have APR 1.3+, we can unbuffer the file to let us use
147 * mmap and zero-copy the PUT body. However, on older APR versions,
148 * we can't check the buffer status; but serf will fall through and
149 * create a file bucket for us on the buffered handle.
150 */
151
152 SVN_ERR(svn_io_file_flush(b->file, b->scratch_pool));
153 apr_file_buffer_set(b->file, NULL, 0);
154 }
155 else if (b->collect_bucket)
156 b->all_data = allocate_all(b, b->result_pool);
157
158 if (b->scratch_pool)
159 svn_pool_destroy(b->scratch_pool);
160
161 return SVN_NO_ERROR;
162 }
163
164 /* Implements svn_ra_serf__request_body_delegate_t. */
165 static svn_error_t *
request_body_delegate(serf_bucket_t ** body_bkt,void * baton,serf_bucket_alloc_t * alloc,apr_pool_t * request_pool,apr_pool_t * scratch_pool)166 request_body_delegate(serf_bucket_t **body_bkt,
167 void *baton,
168 serf_bucket_alloc_t *alloc,
169 apr_pool_t *request_pool,
170 apr_pool_t *scratch_pool)
171 {
172 svn_ra_serf__request_body_t *b = baton;
173
174 if (b->file)
175 {
176 apr_off_t offset;
177
178 offset = 0;
179 SVN_ERR(svn_io_file_seek(b->file, APR_SET, &offset, scratch_pool));
180
181 *body_bkt = serf_bucket_file_create(b->file, alloc);
182 }
183 else
184 {
185 *body_bkt = serf_bucket_simple_create(b->all_data,
186 b->total_bytes,
187 NULL, NULL, alloc);
188 }
189
190 return SVN_NO_ERROR;
191 }
192
193 svn_ra_serf__request_body_t *
svn_ra_serf__request_body_create(apr_size_t in_memory_size,apr_pool_t * result_pool)194 svn_ra_serf__request_body_create(apr_size_t in_memory_size,
195 apr_pool_t *result_pool)
196 {
197 svn_ra_serf__request_body_t *body = apr_pcalloc(result_pool, sizeof(*body));
198
199 body->in_memory_size = in_memory_size;
200 body->result_pool = result_pool;
201 body->stream = svn_stream_create(body, result_pool);
202
203 svn_stream_set_write(body->stream, request_body_stream_write);
204 svn_stream_set_close(body->stream, request_body_stream_close);
205
206 return body;
207 }
208
209 svn_stream_t *
svn_ra_serf__request_body_get_stream(svn_ra_serf__request_body_t * body)210 svn_ra_serf__request_body_get_stream(svn_ra_serf__request_body_t *body)
211 {
212 return body->stream;
213 }
214
215 void
svn_ra_serf__request_body_get_delegate(svn_ra_serf__request_body_delegate_t * del,void ** baton,svn_ra_serf__request_body_t * body)216 svn_ra_serf__request_body_get_delegate(svn_ra_serf__request_body_delegate_t *del,
217 void **baton,
218 svn_ra_serf__request_body_t *body)
219 {
220 *del = request_body_delegate;
221 *baton = body;
222 }
223
224 svn_error_t *
svn_ra_serf__request_body_cleanup(svn_ra_serf__request_body_t * body,apr_pool_t * scratch_pool)225 svn_ra_serf__request_body_cleanup(svn_ra_serf__request_body_t *body,
226 apr_pool_t *scratch_pool)
227 {
228 if (body->file)
229 SVN_ERR(svn_io_file_close(body->file, scratch_pool));
230
231 return SVN_NO_ERROR;
232 }
233