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
28struct 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. */
43static const char *
44allocate_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. */
76static void serf_free_no_error(void *unfreed_baton, void *block) {}
77
78/* Stream write function for body creation. */
79static svn_error_t *
80request_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. */
135static svn_error_t *
136request_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. */
165static svn_error_t *
166request_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
193svn_ra_serf__request_body_t *
194svn_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
209svn_stream_t *
210svn_ra_serf__request_body_get_stream(svn_ra_serf__request_body_t *body)
211{
212  return body->stream;
213}
214
215void
216svn_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
224svn_error_t *
225svn_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