subversion/libsvn_subr/stream.c

   1 /*
   2  * stream.c:   svn_stream operations
   3  *
   4  * ====================================================================
   5  * Copyright (c) 2000-2004, 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 #include "svn_private_config.h"
  20
  21 #include <assert.h>
  22 #include <stdio.h>
  23
  24 #include <apr.h>
  25 #include <apr_pools.h>
  26 #include <apr_strings.h>
  27 #include <apr_file_io.h>
  28 #include <apr_errno.h>
  29 #include <apr_md5.h>
  30
  31 #include <zlib.h>
  32
  33 #include "svn_pools.h"
  34 #include "svn_io.h"
  35 #include "svn_error.h"
  36 #include "svn_string.h"
  37 #include "svn_utf.h"
  38
  39
  40 struct svn_stream_t {
  41   void *baton;
  42   svn_read_fn_t read_fn;
  43   svn_write_fn_t write_fn;
  44   svn_close_fn_t close_fn;
  45 };
  46
  47
  48 \f
  49 /*** Generic streams. ***/
  50
  51 svn_stream_t *
  52 svn_stream_create(void *baton, apr_pool_t *pool)
  53 {
  54   svn_stream_t *stream;
  55
  56   stream = apr_palloc(pool, sizeof(*stream));
  57   stream->baton = baton;
  58   stream->read_fn = NULL;
  59   stream->write_fn = NULL;
  60   stream->close_fn = NULL;
  61   return stream;
  62 }
  63
  64
  65 void
  66 svn_stream_set_baton(svn_stream_t *stream, void *baton)
  67 {
  68   stream->baton = baton;
  69 }
  70
  71
  72 void
  73 svn_stream_set_read(svn_stream_t *stream, svn_read_fn_t read_fn)
  74 {
  75   stream->read_fn = read_fn;
  76 }
  77
  78
  79 void
  80 svn_stream_set_write(svn_stream_t *stream, svn_write_fn_t write_fn)
  81 {
  82   stream->write_fn = write_fn;
  83 }
  84
  85
  86 void
  87 svn_stream_set_close(svn_stream_t *stream, svn_close_fn_t close_fn)
  88 {
  89   stream->close_fn = close_fn;
  90 }
  91
  92
  93 svn_error_t *
  94 svn_stream_read(svn_stream_t *stream, char *buffer, apr_size_t *len)
  95 {
  96   assert(stream->read_fn != NULL);
  97   return stream->read_fn(stream->baton, buffer, len);
  98 }
  99
 100
 101 svn_error_t *
 102 svn_stream_write(svn_stream_t *stream, const char *data, apr_size_t *len)
 103 {
 104   assert(stream->write_fn != NULL);
 105   return stream->write_fn(stream->baton, data, len);
 106 }
 107
 108
 109 svn_error_t *
 110 svn_stream_close(svn_stream_t *stream)
 111 {
 112   if (stream->close_fn == NULL)
 113     return SVN_NO_ERROR;
 114   return stream->close_fn(stream->baton);
 115 }
 116
 117
 118 svn_error_t *
 119 svn_stream_printf(svn_stream_t *stream,
 120                   apr_pool_t *pool,
 121                   const char *fmt,
 122                   ...)
 123 {
 124   const char *message;
 125   va_list ap;
 126   apr_size_t len;
 127
 128   va_start(ap, fmt);
 129   message = apr_pvsprintf(pool, fmt, ap);
 130   va_end(ap);
 131
 132   len = strlen(message);
 133   return svn_stream_write(stream, message, &len);
 134 }
 135
 136
 137 svn_error_t *
 138 svn_stream_printf_from_utf8(svn_stream_t *stream,
 139                             const char *encoding,
 140                             apr_pool_t *pool,
 141                             const char *fmt,
 142                             ...)
 143 {
 144   const char *message, *translated;
 145   va_list ap;
 146   apr_size_t len;
 147
 148   va_start(ap, fmt);
 149   message = apr_pvsprintf(pool, fmt, ap);
 150   va_end(ap);
 151
 152   SVN_ERR(svn_utf_cstring_from_utf8_ex2(&translated, message, encoding,
 153                                         pool));
 154
 155   len = strlen(translated);
 156
 157   return svn_stream_write(stream, translated, &len);
 158 }
 159
 160
 161 svn_error_t *
 162 svn_stream_readline(svn_stream_t *stream,
 163                     svn_stringbuf_t **stringbuf,
 164                     const char *eol,
 165                     svn_boolean_t *eof,
 166                     apr_pool_t *pool)
 167 {
 168   apr_size_t numbytes;
 169   const char *match;
 170   char c;
 171   /* Since we're reading one character at a time, let's at least
 172      optimize for the 90% case.  90% of the time, we can avoid the
 173      stringbuf ever having to realloc() itself if we start it out at
 174      80 chars.  */
 175   svn_stringbuf_t *str = svn_stringbuf_create_ensure(80, pool);
 176
 177   match = eol;
 178   while (*match)
 179     {
 180       numbytes = 1;
 181       SVN_ERR(svn_stream_read(stream, &c, &numbytes));
 182       if (numbytes != 1)
 183         {
 184           /* a 'short' read means the stream has run out. */
 185           *eof = TRUE;
 186           *stringbuf = str;
 187           return SVN_NO_ERROR;
 188         }
 189
 190       if (c == *match)
 191         match++;
 192       else
 193         match = eol;
 194
 195       svn_stringbuf_appendbytes(str, &c, 1);
 196     }
 197
 198   *eof = FALSE;
 199   svn_stringbuf_chop(str, match - eol);
 200   *stringbuf = str;
 201   return SVN_NO_ERROR;
 202 }
 203
 204
 205 svn_error_t *svn_stream_copy2(svn_stream_t *from, svn_stream_t *to,
 206                               svn_cancel_func_t cancel_func,
 207                               void *cancel_baton,
 208                               apr_pool_t *pool)
 209 {
 210   char *buf = apr_palloc(pool, SVN__STREAM_CHUNK_SIZE);
 211   apr_size_t len;
 212
 213   /* Read and write chunks until we get a short read, indicating the
 214      end of the stream.  (We can't get a short write without an
 215      associated error.) */
 216   while (1)
 217     {
 218       len = SVN__STREAM_CHUNK_SIZE;
 219
 220       if (cancel_func)
 221         SVN_ERR(cancel_func(cancel_baton));
 222
 223       SVN_ERR(svn_stream_read(from, buf, &len));
 224       if (len > 0)
 225         SVN_ERR(svn_stream_write(to, buf, &len));
 226       if (len != SVN__STREAM_CHUNK_SIZE)
 227         break;
 228     }
 229   return SVN_NO_ERROR;
 230 }
 231
 232 svn_error_t *svn_stream_copy(svn_stream_t *from, svn_stream_t *to,
 233                              apr_pool_t *pool)
 234 {
 235   return svn_stream_copy2(from, to, NULL, NULL, pool);
 236 }
 237
 238 svn_error_t *
 239 svn_stream_contents_same(svn_boolean_t *same,
 240                          svn_stream_t *stream1,
 241                          svn_stream_t *stream2,
 242                          apr_pool_t *pool)
 243 {
 244   char *buf1 = apr_palloc(pool, SVN__STREAM_CHUNK_SIZE);
 245   char *buf2 = apr_palloc(pool, SVN__STREAM_CHUNK_SIZE);
 246   apr_size_t bytes_read1 = SVN__STREAM_CHUNK_SIZE;
 247   apr_size_t bytes_read2 = SVN__STREAM_CHUNK_SIZE;
 248
 249   *same = TRUE;  /* assume TRUE, until disproved below */
 250   while (bytes_read1 == SVN__STREAM_CHUNK_SIZE
 251          && bytes_read2 == SVN__STREAM_CHUNK_SIZE)
 252     {
 253       SVN_ERR(svn_stream_read(stream1, buf1, &bytes_read1));
 254       SVN_ERR(svn_stream_read(stream2, buf2, &bytes_read2));
 255
 256       if ((bytes_read1 != bytes_read2)
 257           || (memcmp(buf1, buf2, bytes_read1)))
 258         {
 259           *same = FALSE;
 260           break;
 261         }
 262     }
 263
 264   return SVN_NO_ERROR;
 265 }
 266
 267
 268 \f
 269 /*** Generic readable empty stream ***/
 270
 271 static svn_error_t *
 272 read_handler_empty(void *baton, char *buffer, apr_size_t *len)
 273 {
 274   *len = 0;
 275   return SVN_NO_ERROR;
 276 }
 277
 278
 279 static svn_error_t *
 280 write_handler_empty(void *baton, const char *data, apr_size_t *len)
 281 {
 282   return SVN_NO_ERROR;
 283 }
 284
 285
 286 svn_stream_t *
 287 svn_stream_empty(apr_pool_t *pool)
 288 {
 289   svn_stream_t *stream;
 290
 291   stream = svn_stream_create(NULL, pool);
 292   svn_stream_set_read(stream, read_handler_empty);
 293   svn_stream_set_write(stream, write_handler_empty);
 294   return stream;
 295 }
 296
 297
 298
 299 \f
 300 /*** Ownership detaching stream ***/
 301
 302 static svn_error_t *
 303 read_handler_disown(void *baton, char *buffer, apr_size_t *len)
 304 {
 305   return svn_stream_read((svn_stream_t *)baton, buffer, len);
 306 }
 307
 308 static svn_error_t *
 309 write_handler_disown(void *baton, const char *buffer, apr_size_t *len)
 310 {
 311   return svn_stream_write((svn_stream_t *)baton, buffer, len);
 312 }
 313
 314
 315 svn_stream_t *
 316 svn_stream_disown(svn_stream_t *stream, apr_pool_t *pool)
 317 {
 318   svn_stream_t *s = svn_stream_create(stream, pool);
 319
 320   svn_stream_set_read(s, read_handler_disown);
 321   svn_stream_set_write(s, write_handler_disown);
 322
 323   return s;
 324 }
 325
 326
 327 \f
 328 /*** Generic stream for APR files ***/
 329 struct baton_apr {
 330   apr_file_t *file;
 331   apr_pool_t *pool;
 332 };
 333
 334
 335 static svn_error_t *
 336 read_handler_apr(void *baton, char *buffer, apr_size_t *len)
 337 {
 338   struct baton_apr *btn = baton;
 339   svn_error_t *err;
 340
 341   err = svn_io_file_read_full(btn->file, buffer, *len, len, btn->pool);
 342   if (err && APR_STATUS_IS_EOF(err->apr_err))
 343     {
 344       svn_error_clear(err);
 345       err = SVN_NO_ERROR;
 346     }
 347
 348   return err;
 349 }
 350
 351
 352 static svn_error_t *
 353 write_handler_apr(void *baton, const char *data, apr_size_t *len)
 354 {
 355   struct baton_apr *btn = baton;
 356
 357   return svn_io_file_write_full(btn->file, data, *len, len, btn->pool);
 358 }
 359
 360 static svn_error_t *
 361 close_handler_apr(void *baton)
 362 {
 363   struct baton_apr *btn = baton;
 364
 365   return svn_io_file_close(btn->file, btn->pool);
 366 }
 367
 368
 369 svn_stream_t *
 370 svn_stream_from_aprfile2(apr_file_t *file,
 371                          svn_boolean_t disown,
 372                          apr_pool_t *pool)
 373 {
 374   struct baton_apr *baton;
 375   svn_stream_t *stream;
 376
 377   if (file == NULL)
 378     return svn_stream_empty(pool);
 379
 380   baton = apr_palloc(pool, sizeof(*baton));
 381   baton->file = file;
 382   baton->pool = pool;
 383   stream = svn_stream_create(baton, pool);
 384   svn_stream_set_read(stream, read_handler_apr);
 385   svn_stream_set_write(stream, write_handler_apr);
 386
 387   if (! disown)
 388     svn_stream_set_close(stream, close_handler_apr);
 389
 390   return stream;
 391 }
 392
 393 svn_stream_t *
 394 svn_stream_from_aprfile(apr_file_t *file, apr_pool_t *pool)
 395 {
 396   return svn_stream_from_aprfile2(file, TRUE, pool);
 397 }
 398
 399
 400 \f
 401 /* Compressed stream support */
 402
 403 #define ZBUFFER_SIZE 4096       /* The size of the buffer the
 404                                    compressed stream uses to read from
 405                                    the substream. Basically an
 406                                    arbitrary value, picked to be about
 407                                    page-sized. */
 408
 409 struct zbaton {
 410   z_stream *in;                 /* compressed stream for reading */
 411   z_stream *out;                /* compressed stream for writing */
 412   svn_read_fn_t read;           /* substream's read function */
 413   svn_write_fn_t write;         /* substream's write function */
 414   svn_close_fn_t close;         /* substream's close function */
 415   void *read_buffer;            /* buffer   used   for  reading   from
 416                                    substream */
 417   int read_flush;               /* what flush mode to use while
 418                                    reading */
 419   apr_pool_t *pool;             /* The pool this baton is allocated
 420                                    on */
 421   void *subbaton;               /* The substream's baton */
 422 };
 423
 424 /* zlib alloc function. opaque is the pool we need. */
 425 static voidpf
 426 zalloc(voidpf opaque, uInt items, uInt size)
 427 {
 428   apr_pool_t *pool = opaque;
 429
 430   return apr_palloc(pool, items * size);
 431 }
 432
 433 /* zlib free function */
 434 static void
 435 zfree(voidpf opaque, voidpf address)
 436 {
 437   /* Empty, since we allocate on the pool */
 438 }
 439
 440 /* Converts a zlib error to an svn_error_t. zerr is the error code,
 441    function is the function name, and stream is the z_stream we are
 442    using.  */
 443 static svn_error_t *
 444 zerr_to_svn_error(int zerr, const char *function, z_stream *stream)
 445 {
 446   apr_status_t status;
 447   const char *message;
 448
 449   if (zerr == Z_OK)
 450     return SVN_NO_ERROR;
 451
 452   switch (zerr)
 453     {
 454     case Z_STREAM_ERROR:
 455       status = SVN_ERR_STREAM_MALFORMED_DATA;
 456       message = "stream error";
 457       break;
 458
 459     case Z_MEM_ERROR:
 460       status = APR_ENOMEM;
 461       message = "out of memory";
 462       break;
 463
 464     case Z_BUF_ERROR:
 465       status = APR_ENOMEM;
 466       message = "buffer error";
 467       break;
 468
 469     case Z_VERSION_ERROR:
 470       status = SVN_ERR_STREAM_UNRECOGNIZED_DATA;
 471       message = "version error";
 472       break;
 473
 474     case Z_DATA_ERROR:
 475       status = SVN_ERR_STREAM_MALFORMED_DATA;
 476       message = "corrupted data";
 477       break;
 478
 479     default:
 480       status = SVN_ERR_STREAM_UNRECOGNIZED_DATA;
 481       message = "error";
 482       break;
 483     }
 484
 485   if (stream->msg != NULL)
 486     return svn_error_createf(status, NULL, "zlib (%s): %s: %s", function,
 487                              message, stream->msg);
 488   else
 489     return svn_error_createf(status, NULL, "zlib (%s): %s", function,
 490                              message);
 491 }
 492
 493 /* Helper function to figure out the sync mode */
 494 static svn_error_t *
 495 read_helper_gz(svn_read_fn_t read_fn,
 496                void *baton,
 497                char *buffer,
 498                uInt *len, int *zflush)
 499 {
 500   uInt orig_len = *len;
 501
 502   /* There's no reason this value should grow bigger than the range of
 503      uInt, but Subversion's API requires apr_size_t. */
 504   apr_size_t apr_len = (apr_size_t) *len;
 505
 506   SVN_ERR((*read_fn)(baton, buffer, &apr_len));
 507
 508   /* Type cast back to uInt type that zlib uses.  On LP64 platforms
 509      apr_size_t will be bigger than uInt. */
 510   *len = (uInt) apr_len;
 511
 512   /* I wanted to use Z_FINISH here, but we need to know our buffer is
 513      big enough */
 514   *zflush = (*len) < orig_len ? Z_SYNC_FLUSH : Z_SYNC_FLUSH;
 515
 516   return SVN_NO_ERROR;
 517 }
 518
 519 /* Handle reading from a compressed stream */
 520 static svn_error_t *
 521 read_handler_gz(void *baton, char *buffer, apr_size_t *len)
 522 {
 523   struct zbaton *btn = baton;
 524   int zerr;
 525
 526   if (btn->in == NULL)
 527     {
 528       btn->in = apr_palloc(btn->pool, sizeof(z_stream));
 529       btn->in->zalloc = zalloc;
 530       btn->in->zfree = zfree;
 531       btn->in->opaque = btn->pool;
 532       btn->read_buffer = apr_palloc(btn->pool, ZBUFFER_SIZE);
 533       btn->in->next_in = btn->read_buffer;
 534       btn->in->avail_in = ZBUFFER_SIZE;
 535
 536       SVN_ERR(read_helper_gz(btn->read, btn->subbaton, btn->read_buffer,
 537                              &btn->in->avail_in, &btn->read_flush));
 538
 539       zerr = inflateInit(btn->in);
 540       SVN_ERR(zerr_to_svn_error(zerr, "inflateInit", btn->in));
 541     }
 542
 543   btn->in->next_out = (Bytef *) buffer;
 544   btn->in->avail_out = *len;
 545
 546   while (btn->in->avail_out > 0)
 547     {
 548       if (btn->in->avail_in <= 0)
 549         {
 550           btn->in->avail_in = ZBUFFER_SIZE;
 551           btn->in->next_in = btn->read_buffer;
 552           SVN_ERR(read_helper_gz(btn->read, btn->subbaton, btn->read_buffer,
 553                                  &btn->in->avail_in, &btn->read_flush));
 554         }
 555
 556       zerr = inflate(btn->in, btn->read_flush);
 557       if (zerr == Z_STREAM_END)
 558         break;
 559       else if (zerr != Z_OK)
 560         return zerr_to_svn_error(zerr, "inflate", btn->in);
 561     }
 562
 563   *len -= btn->in->avail_out;
 564   return SVN_NO_ERROR;
 565 }
 566
 567 /* Compress data and write it to the substream */
 568 static svn_error_t *
 569 write_handler_gz(void *baton, const char *buffer, apr_size_t *len)
 570 {
 571   struct zbaton *btn = baton;
 572   apr_pool_t *subpool;
 573   void *write_buf;
 574   apr_size_t buf_size, write_len;
 575   int zerr;
 576
 577   if (btn->out == NULL)
 578     {
 579       btn->out = apr_palloc(btn->pool, sizeof(z_stream));
 580       btn->out->zalloc = zalloc;
 581       btn->out->zfree = zfree;
 582       btn->out->opaque =  btn->pool;
 583
 584       zerr = deflateInit(btn->out, Z_DEFAULT_COMPRESSION);
 585       SVN_ERR(zerr_to_svn_error(zerr, "deflateInit", btn->out));
 586     }
 587
 588   /* The largest buffer we should need is 0.1% larger than the
 589      compressed data, + 12 bytes. This info comes from zlib.h.  */
 590   buf_size = *len + (*len / 1000) + 13;
 591   subpool = svn_pool_create(btn->pool);
 592   write_buf = apr_palloc(subpool, buf_size);
 593
 594   btn->out->next_in = (Bytef *) buffer;  /* Casting away const! */
 595   btn->out->avail_in = *len;
 596
 597   while (btn->out->avail_in > 0)
 598     {
 599       btn->out->next_out = write_buf;
 600       btn->out->avail_out = buf_size;
 601
 602       zerr = deflate(btn->out, Z_NO_FLUSH);
 603       SVN_ERR(zerr_to_svn_error(zerr, "deflate", btn->out));
 604       write_len = buf_size - btn->out->avail_out;
 605       if (write_len > 0)
 606         SVN_ERR(btn->write(btn->subbaton, write_buf, &write_len));
 607     }
 608
 609   svn_pool_destroy(subpool);
 610
 611   return SVN_NO_ERROR;
 612 }
 613
 614 /* Handle flushing and closing the stream */
 615 static svn_error_t *
 616 close_handler_gz(void *baton)
 617 {
 618   struct zbaton *btn = baton;
 619   int zerr;
 620
 621   if (btn->in != NULL)
 622     {
 623       zerr = inflateEnd(btn->in);
 624       SVN_ERR(zerr_to_svn_error(zerr, "inflateEnd", btn->in));
 625     }
 626
 627   if (btn->out != NULL)
 628     {
 629       void *buf;
 630       apr_size_t write_len;
 631
 632       buf = apr_palloc(btn->pool, ZBUFFER_SIZE);
 633
 634       while (TRUE)
 635         {
 636           btn->out->next_out = buf;
 637           btn->out->avail_out = ZBUFFER_SIZE;
 638
 639           zerr = deflate(btn->out, Z_FINISH);
 640           if (zerr != Z_STREAM_END && zerr != Z_OK)
 641             return zerr_to_svn_error(zerr, "deflate", btn->out);
 642           write_len = ZBUFFER_SIZE - btn->out->avail_out;
 643           if (write_len > 0)
 644             SVN_ERR(btn->write(btn->subbaton, buf, &write_len));
 645           if (zerr == Z_STREAM_END)
 646             break;
 647         }
 648
 649       zerr = deflateEnd(btn->out);
 650       SVN_ERR(zerr_to_svn_error(zerr, "deflateEnd", btn->out));
 651     }
 652
 653   if (btn->close != NULL)
 654     return btn->close(btn->subbaton);
 655   else
 656     return SVN_NO_ERROR;
 657 }
 658
 659
 660 svn_stream_t *
 661 svn_stream_compressed(svn_stream_t *stream, apr_pool_t *pool)
 662 {
 663   struct svn_stream_t *zstream;
 664   struct zbaton *baton;
 665
 666   assert(stream != NULL);
 667
 668   baton = apr_palloc(pool, sizeof(*baton));
 669   baton->in = baton->out = NULL;
 670   baton->read = stream->read_fn;
 671   baton->write = stream->write_fn;
 672   baton->close = stream->close_fn;
 673   baton->subbaton = stream->baton;
 674   baton->pool = pool;
 675   baton->read_buffer = NULL;
 676   baton->read_flush = Z_SYNC_FLUSH;
 677
 678   zstream = svn_stream_create(baton, pool);
 679   svn_stream_set_read(zstream, read_handler_gz);
 680   svn_stream_set_write(zstream, write_handler_gz);
 681   svn_stream_set_close(zstream, close_handler_gz);
 682
 683   return zstream;
 684 }
 685
 686 \f
 687 /* MD5 checked stream support */
 688
 689 struct md5_stream_baton
 690 {
 691   apr_md5_ctx_t read_ctx, write_ctx;
 692   const unsigned char **read_digest;
 693   const unsigned char **write_digest;
 694   unsigned char read_digest_buf[APR_MD5_DIGESTSIZE];
 695   unsigned char write_digest_buf[APR_MD5_DIGESTSIZE];
 696   svn_stream_t *proxy;
 697
 698   /* True if more data should be read when closing the stream. */
 699   svn_boolean_t read_more;
 700
 701   /* Pool to allocate read buffer from. */
 702   apr_pool_t *pool;
 703 };
 704
 705 static svn_error_t *
 706 read_handler_md5(void *baton, char *buffer, apr_size_t *len)
 707 {
 708   struct md5_stream_baton *btn = baton;
 709   apr_size_t saved_len = *len;
 710
 711   SVN_ERR(svn_stream_read(btn->proxy, buffer, len));
 712
 713   if (btn->read_digest)
 714     {
 715       apr_status_t apr_err = apr_md5_update(&btn->read_ctx, buffer, *len);
 716
 717       if (apr_err)
 718         return svn_error_create(apr_err, NULL, NULL);
 719     }
 720
 721   if (saved_len != *len)
 722     btn->read_more = FALSE;
 723
 724   return SVN_NO_ERROR;
 725 }
 726
 727
 728 static svn_error_t *
 729 write_handler_md5(void *baton, const char *buffer, apr_size_t *len)
 730 {
 731   struct md5_stream_baton *btn = baton;
 732
 733   if (btn->write_digest && *len > 0)
 734     {
 735       apr_status_t apr_err = apr_md5_update(&btn->write_ctx, buffer, *len);
 736
 737       if (apr_err)
 738         return svn_error_create(apr_err, NULL, NULL);
 739     }
 740
 741   return svn_stream_write(btn->proxy, buffer, len);
 742 }
 743
 744
 745 static svn_error_t *
 746 close_handler_md5(void *baton)
 747 {
 748   struct md5_stream_baton *btn = baton;
 749
 750   /* If we're supposed to drain the stream, do so before finalizing the
 751      checksum. */
 752   if (btn->read_more)
 753     {
 754       char *buf = apr_palloc(btn->pool, SVN__STREAM_CHUNK_SIZE);
 755       apr_size_t len = SVN__STREAM_CHUNK_SIZE;
 756
 757       do
 758         {
 759           SVN_ERR(read_handler_md5(baton, buf, &len));
 760         }
 761       while (btn->read_more);
 762     }
 763
 764   if (btn->read_digest)
 765     {
 766       apr_status_t apr_err
 767         = apr_md5_final(btn->read_digest_buf, &btn->read_ctx);
 768
 769       if (apr_err)
 770         return svn_error_create(apr_err, NULL, NULL);
 771
 772       *btn->read_digest = btn->read_digest_buf;
 773     }
 774
 775   if (btn->write_digest)
 776     {
 777       apr_status_t apr_err
 778         = apr_md5_final(btn->write_digest_buf, &btn->write_ctx);
 779
 780       if (apr_err)
 781         return svn_error_create(apr_err, NULL, NULL);
 782
 783       *btn->write_digest = btn->write_digest_buf;
 784     }
 785
 786   return svn_stream_close(btn->proxy);
 787 }
 788
 789
 790 svn_stream_t *
 791 svn_stream_checksummed(svn_stream_t *stream,
 792                        const unsigned char **read_digest,
 793                        const unsigned char **write_digest,
 794                        svn_boolean_t read_all,
 795                        apr_pool_t *pool)
 796 {
 797   svn_stream_t *s;
 798   struct md5_stream_baton *baton;
 799
 800   if (! read_digest && ! write_digest)
 801     return stream;
 802
 803   baton = apr_palloc(pool, sizeof(*baton));
 804   apr_md5_init(&baton->read_ctx);
 805   apr_md5_init(&baton->write_ctx);
 806   baton->read_digest = read_digest;
 807   baton->write_digest = write_digest;
 808   baton->proxy = stream;
 809   baton->read_more = read_all;
 810   baton->pool = pool;
 811
 812   s = svn_stream_create(baton, pool);
 813   svn_stream_set_read(s, read_handler_md5);
 814   svn_stream_set_write(s, write_handler_md5);
 815   svn_stream_set_close(s, close_handler_md5);
 816   return s;
 817 }
 818
 819
 820
 821 \f
 822 /* Miscellaneous stream functions. */
 823 struct string_stream_baton
 824 {
 825   svn_stringbuf_t *str;
 826   apr_size_t amt_read;
 827 };
 828
 829 static svn_error_t *
 830 read_handler_string(void *baton, char *buffer, apr_size_t *len)
 831 {
 832   struct string_stream_baton *btn = baton;
 833   apr_size_t left_to_read = btn->str->len - btn->amt_read;
 834
 835   *len = (*len > left_to_read) ? left_to_read : *len;
 836   memcpy(buffer, btn->str->data + btn->amt_read, *len);
 837   btn->amt_read += *len;
 838   return SVN_NO_ERROR;
 839 }
 840
 841 static svn_error_t *
 842 write_handler_string(void *baton, const char *data, apr_size_t *len)
 843 {
 844   struct string_stream_baton *btn = baton;
 845
 846   svn_stringbuf_appendbytes(btn->str, data, *len);
 847   return SVN_NO_ERROR;
 848 }
 849
 850 svn_stream_t *
 851 svn_stream_from_stringbuf(svn_stringbuf_t *str,
 852                           apr_pool_t *pool)
 853 {
 854   svn_stream_t *stream;
 855   struct string_stream_baton *baton;
 856
 857   if (! str)
 858     return svn_stream_empty(pool);
 859
 860   baton = apr_palloc(pool, sizeof(*baton));
 861   baton->str = str;
 862   baton->amt_read = 0;
 863   stream = svn_stream_create(baton, pool);
 864   svn_stream_set_read(stream, read_handler_string);
 865   svn_stream_set_write(stream, write_handler_string);
 866   return stream;
 867 }
 868
 869
 870 svn_error_t *
 871 svn_stream_for_stdout(svn_stream_t **out, apr_pool_t *pool)
 872 {
 873   apr_file_t *stdout_file;
 874   apr_status_t apr_err;
 875
 876   apr_err = apr_file_open_stdout(&stdout_file, pool);
 877   if (apr_err)
 878     return svn_error_wrap_apr(apr_err, "Can't open stdout");
 879
 880   *out = svn_stream_from_aprfile(stdout_file, pool);
 881
 882   return SVN_NO_ERROR;
 883 }