subversion/libsvn_ra_svn/streams.c

   1 /*
   2  * streams.c :  stream encapsulation routines for the ra_svn protocol
   3  *
   4  * ====================================================================
   5  * Copyright (c) 2000-2006 CollabNet.  All rights reserved.
   6  *
   7  * This software is licensed as described in the file COPYING, which
   8  * you should have received as part of this distribution.  The terms
   9  * are also available at http://subversion.tigris.org/license-1.html.
  10  * If newer versions of this license are posted there, you may use a
  11  * newer version instead, at your option.
  12  *
  13  * This software consists of voluntary contributions made by many
  14  * individuals.  For exact contribution history, see the revision
  15  * history and logs, available at http://subversion.tigris.org/.
  16  * ====================================================================
  17  */
  18
  19
  20 \f
  21 #include <apr_general.h>
  22 #include <apr_network_io.h>
  23 #include <apr_poll.h>
  24
  25 #include "svn_types.h"
  26 #include "svn_error.h"
  27 #include "svn_pools.h"
  28 #include "svn_io.h"
  29 #include "svn_private_config.h"
  30
  31 #include "ra_svn.h"
  32
  33 struct svn_ra_svn__stream_st {
  34   svn_stream_t *stream;
  35   void *baton;
  36   ra_svn_pending_fn_t pending_fn;
  37   ra_svn_timeout_fn_t timeout_fn;
  38 };
  39
  40 typedef struct {
  41   apr_socket_t *sock;
  42   apr_pool_t *pool;
  43 } sock_baton_t;
  44
  45 typedef struct {
  46   apr_file_t *in_file;
  47   apr_file_t *out_file;
  48   apr_pool_t *pool;
  49 } file_baton_t;
  50
  51 /* Returns TRUE if PFD has pending data, FALSE otherwise. */
  52 static svn_boolean_t pending(apr_pollfd_t *pfd, apr_pool_t *pool)
  53 {
  54   apr_status_t status;
  55   int n;
  56
  57   pfd->p = pool;
  58   pfd->reqevents = APR_POLLIN;
  59 #ifdef AS400
  60   status = apr_poll(pfd, 1, &n, 0, pool);
  61 #else
  62   status = apr_poll(pfd, 1, &n, 0);
  63 #endif
  64   return (status == APR_SUCCESS && n);
  65 }
  66
  67 /* Functions to implement a file backed svn_ra_svn__stream_t. */
  68
  69 /* Implements svn_read_fn_t */
  70 static svn_error_t *
  71 file_read_cb(void *baton, char *buffer, apr_size_t *len)
  72 {
  73   file_baton_t *b = baton;
  74   apr_status_t status = apr_file_read(b->in_file, buffer, len);
  75
  76   if (status && !APR_STATUS_IS_EOF(status))
  77     return svn_error_wrap_apr(status, _("Can't read from connection"));
  78   if (*len == 0)
  79     return svn_error_create(SVN_ERR_RA_SVN_CONNECTION_CLOSED, NULL,
  80                             _("Connection closed unexpectedly"));
  81   return SVN_NO_ERROR;
  82 }
  83
  84 /* Implements svn_write_fn_t */
  85 static svn_error_t *
  86 file_write_cb(void *baton, const char *buffer, apr_size_t *len)
  87 {
  88   file_baton_t *b = baton;
  89   apr_status_t status = apr_file_write(b->out_file, buffer, len);
  90   if (status)
  91     return svn_error_wrap_apr(status, _("Can't write to connection"));
  92   return SVN_NO_ERROR;
  93 }
  94
  95 /* Implements ra_svn_timeout_fn_t */
  96 static void
  97 file_timeout_cb(void *baton, apr_interval_time_t interval)
  98 {
  99   file_baton_t *b = baton;
 100   apr_file_pipe_timeout_set(b->out_file, interval);
 101 }
 102
 103 /* Implements ra_svn_pending_fn_t */
 104 static svn_boolean_t
 105 file_pending_cb(void *baton)
 106 {
 107   file_baton_t *b = baton;
 108   apr_pollfd_t pfd;
 109
 110   pfd.desc_type = APR_POLL_FILE;
 111   pfd.desc.f = b->in_file;
 112
 113   return pending(&pfd, b->pool);
 114 }
 115
 116 svn_ra_svn__stream_t *
 117 svn_ra_svn__stream_from_files(apr_file_t *in_file,
 118                               apr_file_t *out_file,
 119                               apr_pool_t *pool)
 120 {
 121   file_baton_t *b = apr_palloc(pool, sizeof(*b));
 122
 123   b->in_file = in_file;
 124   b->out_file = out_file;
 125   b->pool = pool;
 126
 127   return svn_ra_svn__stream_create(b, file_read_cb, file_write_cb,
 128                                    file_timeout_cb, file_pending_cb,
 129                                    pool);
 130 }
 131
 132 /* Functions to implement a socket backed svn_ra_svn__stream_t. */
 133
 134 /* Implements svn_read_fn_t */
 135 static svn_error_t *
 136 sock_read_cb(void *baton, char *buffer, apr_size_t *len)
 137 {
 138   sock_baton_t *b = baton;
 139   apr_status_t status;
 140   apr_interval_time_t interval;
 141
 142   status = apr_socket_timeout_get(b->sock, &interval);
 143   if (status)
 144     return svn_error_wrap_apr(status, _("Can't get socket timeout"));
 145
 146   /* Always block on read. */
 147   apr_socket_timeout_set(b->sock, -1);
 148   status = apr_socket_recv(b->sock, buffer, len);
 149   apr_socket_timeout_set(b->sock, interval);
 150
 151   if (status && !APR_STATUS_IS_EOF(status))
 152     return svn_error_wrap_apr(status, _("Can't read from connection"));
 153   if (*len == 0)
 154     return svn_error_create(SVN_ERR_RA_SVN_CONNECTION_CLOSED, NULL,
 155                             _("Connection closed unexpectedly"));
 156   return SVN_NO_ERROR;
 157 }
 158
 159 /* Implements svn_write_fn_t */
 160 static svn_error_t *
 161 sock_write_cb(void *baton, const char *buffer, apr_size_t *len)
 162 {
 163   sock_baton_t *b = baton;
 164   apr_status_t status = apr_socket_send(b->sock, buffer, len);
 165   if (status)
 166     return svn_error_wrap_apr(status, _("Can't write to connection"));
 167   return SVN_NO_ERROR;
 168 }
 169
 170 /* Implements ra_svn_timeout_fn_t */
 171 static void
 172 sock_timeout_cb(void *baton, apr_interval_time_t interval)
 173 {
 174   sock_baton_t *b = baton;
 175   apr_socket_timeout_set(b->sock, interval);
 176 }
 177
 178 /* Implements ra_svn_pending_fn_t */
 179 static svn_boolean_t
 180 sock_pending_cb(void *baton)
 181 {
 182   sock_baton_t *b = baton;
 183   apr_pollfd_t pfd;
 184
 185   pfd.desc_type = APR_POLL_SOCKET;
 186   pfd.desc.s = b->sock;
 187
 188   return pending(&pfd, b->pool);
 189 }
 190
 191 svn_ra_svn__stream_t *
 192 svn_ra_svn__stream_from_sock(apr_socket_t *sock,
 193                              apr_pool_t *pool)
 194 {
 195   sock_baton_t *b = apr_palloc(pool, sizeof(*b));
 196
 197   b->sock = sock;
 198   b->pool = pool;
 199
 200   return svn_ra_svn__stream_create(b, sock_read_cb, sock_write_cb,
 201                                    sock_timeout_cb, sock_pending_cb,
 202                                    pool);
 203 }
 204
 205 svn_ra_svn__stream_t *
 206 svn_ra_svn__stream_create(void *baton,
 207                           svn_read_fn_t read_cb,
 208                           svn_write_fn_t write_cb,
 209                           ra_svn_timeout_fn_t timeout_cb,
 210                           ra_svn_pending_fn_t pending_cb,
 211                           apr_pool_t *pool)
 212 {
 213   svn_ra_svn__stream_t *s = apr_palloc(pool, sizeof(*s));
 214   s->stream = svn_stream_empty(pool);
 215   svn_stream_set_baton(s->stream, baton);
 216   if (read_cb)
 217     svn_stream_set_read(s->stream, read_cb);
 218   if (write_cb)
 219     svn_stream_set_write(s->stream, write_cb);
 220   s->baton = baton;
 221   s->timeout_fn = timeout_cb;
 222   s->pending_fn = pending_cb;
 223   return s;
 224 }
 225
 226 svn_error_t *
 227 svn_ra_svn__stream_write(svn_ra_svn__stream_t *stream,
 228                          const char *data, apr_size_t *len)
 229 {
 230   return svn_stream_write(stream->stream, data, len);
 231 }
 232
 233 svn_error_t *
 234 svn_ra_svn__stream_read(svn_ra_svn__stream_t *stream, char *data,
 235                         apr_size_t *len)
 236 {
 237   return svn_stream_read(stream->stream, data, len);
 238 }
 239
 240 void
 241 svn_ra_svn__stream_timeout(svn_ra_svn__stream_t *stream,
 242                            apr_interval_time_t interval)
 243 {
 244   stream->timeout_fn(stream->baton, interval);
 245 }
 246
 247 svn_boolean_t
 248 svn_ra_svn__stream_pending(svn_ra_svn__stream_t *stream)
 249 {
 250   return stream->pending_fn(stream->baton);
 251 }