1 /******************************************************************************* 2 3 Copyright: 4 Copyright (c) 2004 Kris Bell. 5 Some parts copyright (c) 2009-2016 dunnhumby Germany GmbH. 6 All rights reserved. 7 8 License: 9 Tango Dual License: 3-Clause BSD License / Academic Free License v3.0. 10 See LICENSE_TANGO.txt for details. 11 12 Version: 13 Mar 2004: Initial release$(BR) 14 Dec 2006: Outback release 15 16 Authors: Kris Bell 17 18 *******************************************************************************/ 19 20 module ocean.io.stream.Buffered; 21 22 import core.stdc.string; 23 24 import ocean.meta.types.Qualifiers; 25 version (unittest) import ocean.core.Test; 26 import ocean.core.Verify; 27 28 public import ocean.io.model.IConduit; 29 30 import ocean.io.device.Conduit; 31 version (unittest) import ocean.io.device.MemoryDevice; 32 33 34 /// Shorthand aliases 35 public alias BufferedInput Bin; 36 /// ditto 37 public alias BufferedOutput Bout; 38 39 40 /******************************************************************************* 41 42 Buffers the flow of data from a upstream input 43 44 A downstream neighbour can locate and use this buffer instead of creating 45 another instance of their own. 46 47 Note: 48 upstream is closer to the source, and downstream is further away 49 50 *******************************************************************************/ 51 52 public class BufferedInput : InputFilter, InputBuffer 53 { 54 /// Clear/flush are the same 55 public alias flush clear; 56 /// Access the source. 57 public alias InputFilter.input input; 58 59 private void[] data; // The raw data buffer. 60 private size_t index; // Current read position. 61 private size_t extent; // Limit of valid content. 62 63 invariant () 64 { 65 assert(this.index <= this.extent); 66 assert(this.extent <= this.data.length); 67 } 68 69 /*************************************************************************** 70 71 Construct a buffer. 72 73 Construct a Buffer upon the provided input stream. 74 75 Params: 76 stream = An input stream. 77 78 ***************************************************************************/ 79 80 public this (InputStream stream) 81 { 82 verify(stream !is null); 83 this(stream, stream.conduit.bufferSize); 84 } 85 86 /*************************************************************************** 87 88 Construct a buffer. 89 90 Construct a Buffer upon the provided input stream. 91 92 Params: 93 stream = An input stream. 94 capacity = Desired buffer capacity. 95 96 ***************************************************************************/ 97 98 public this (InputStream stream, size_t capacity) 99 { 100 this.set(new ubyte[capacity], 0); 101 super(this.source = stream); 102 } 103 104 /*************************************************************************** 105 106 Attempt to share an upstream Buffer, and create an instance 107 where there's not one available. 108 109 If an upstream Buffer instances is visible, it will be shared. 110 Otherwise, a new instance is created based upon the bufferSize 111 exposed by the stream endpoint (conduit). 112 113 Params: 114 stream = An input stream. 115 116 ***************************************************************************/ 117 118 public static InputBuffer create (InputStream stream) 119 { 120 auto source = stream; 121 auto conduit = source.conduit; 122 while (cast(Mutator) source is null) 123 { 124 auto b = cast(InputBuffer) source; 125 if (b) 126 return b; 127 if (source is conduit) 128 break; 129 source = source.input; 130 verify(source !is null); 131 } 132 133 return new BufferedInput(stream, conduit.bufferSize); 134 } 135 136 /*************************************************************************** 137 138 Place more data from the source stream into this buffer, and returns 139 the number of bytes added. 140 141 This does not compress the current buffer content, so consider doing 142 that explicitly. 143 144 Returns: 145 Number of bytes added, which will be Eof when there is no further 146 input available. 147 Zero is also a valid response, meaning no data was actually added. 148 149 ***************************************************************************/ 150 151 public final size_t populate () 152 { 153 return this.writer(&this.input.read); 154 } 155 156 /*************************************************************************** 157 158 Returns: 159 a void[] slice of the buffer from start to end, 160 where end is exclusive. 161 162 ***************************************************************************/ 163 164 public final void[] opSlice (size_t start, size_t end) 165 { 166 verify(start <= this.extent && end <= this.extent && start <= end); 167 return this.data[start .. end]; 168 } 169 170 /*************************************************************************** 171 172 Retrieve the valid content. 173 174 Returns: 175 A void[] slice of the buffer, from the current position up to 176 the limit of valid content. 177 The content remains in the buffer for future extraction. 178 179 ***************************************************************************/ 180 181 public final void[] slice () 182 { 183 return this.data[this.index .. this.extent]; 184 } 185 186 /*************************************************************************** 187 188 Access buffer content. 189 190 Read a slice of data from the buffer, loading from the 191 conduit as necessary. The specified number of bytes is 192 sliced from the buffer, and marked as having been read 193 when the 'eat' parameter is set true. When 'eat' is set 194 false, the read position is not adjusted. 195 196 The slice cannot be larger than the size of the buffer: use method 197 `fill(void[])` instead where you simply want the content copied, 198 or use `conduit.read()` to extract directly from an attached conduit 199 Also if you need to retain the slice, then it should be `.dup`'d 200 before the buffer is compressed or repopulated. 201 202 Params: 203 size = Number of bytes to access. 204 eat = Whether to consume the content or not. 205 206 Returns: 207 The corresponding buffer slice when successful, or 208 null if there's not enough data available (Eof; Eob). 209 210 Examples: 211 --- 212 // create a buffer with some content 213 auto buffer = new Buffer("hello world"); 214 215 // consume everything unread 216 auto slice = buffer.slice(buffer.this.readable()); 217 --- 218 219 ***************************************************************************/ 220 221 public final void[] slice (size_t size, bool eat = true) 222 { 223 if (size > this.readable()) 224 { 225 // make some space? This will try to leave as much content 226 // in the buffer as possible, such that entire records may 227 // be aliased directly from within. 228 if (size > (this.data.length - this.index)) 229 { 230 if (size <= this.data.length) 231 this.compress(); 232 else 233 this.conduit.error("input buffer is empty"); 234 } 235 236 // populate tail of buffer with new content 237 do { 238 if (this.writer(&this.source.read) is Eof) 239 this.conduit.error("end-of-flow whilst reading"); 240 } while (size > this.readable()); 241 } 242 243 auto i = this.index; 244 if (eat) 245 this.index += size; 246 return this.data[i .. i + size]; 247 } 248 249 /*************************************************************************** 250 251 Read directly from this buffer. 252 253 Exposes the raw data buffer at the current _read position. 254 The delegate is provided with a void[] representing the available 255 data, and should return zero to leave the current _read position 256 intact. 257 258 If the delegate consumes data, it should return the number of 259 bytes consumed; or IConduit.Eof to indicate an error. 260 261 Params: 262 dg = Callback to provide buffer access to. 263 264 Returns: 265 the delegate's return value 266 267 ***************************************************************************/ 268 269 public final size_t reader (scope size_t delegate (const(void)[]) dg) 270 { 271 auto count = dg(this.data[this.index .. this.extent]); 272 273 if (count != Eof) 274 { 275 this.index += count; 276 verify(this.index <= this.extent); 277 } 278 return count; 279 } 280 281 /*************************************************************************** 282 283 Write into this buffer. 284 285 Exposes the raw data buffer at the current _write position, 286 The delegate is provided with a void[] representing space 287 available within the buffer at the current _write position. 288 289 The delegate should return the appropriate number of bytes 290 if it writes valid content, or IConduit.Eof on error. 291 292 Params: 293 dg = The callback to provide buffer access to. 294 295 Returns: 296 the delegate return's value. 297 298 ***************************************************************************/ 299 300 public size_t writer (scope size_t delegate (void[]) dg) 301 { 302 auto count = dg(this.data[this.extent .. $]); 303 304 if (count != Eof) 305 { 306 this.extent += count; 307 verify(this.extent <= this.data.length); 308 } 309 return count; 310 } 311 312 /*************************************************************************** 313 314 Transfer content into the provided dst. 315 316 Populates the provided array with content. We try to 317 satisfy the request from the buffer content, and read 318 directly from an attached conduit when the buffer is empty. 319 320 Params: 321 dst = Destination of the content. 322 323 Returns: 324 the number of bytes read, which may be less than `dst.length`. 325 Eof is returned when no further content is available. 326 327 ***************************************************************************/ 328 329 public final override size_t read (void[] dst) 330 { 331 size_t content = this.readable; 332 if (content) 333 { 334 if (content >= dst.length) 335 content = dst.length; 336 337 // transfer buffer content 338 dst[0 .. content] = this.data[this.index .. this.index + content]; 339 this.index += content; 340 } 341 // pathological cases read directly from conduit 342 else if (dst.length > this.data.length) 343 content = this.source.read(dst); 344 else 345 { 346 if (this.writable is 0) 347 this.index = this.extent = 0; // same as clear, without call-chain 348 349 // keep buffer partially populated 350 if ((content = this.writer(&this.source.read)) != Eof && content > 0) 351 content = this.read(dst); 352 } 353 return content; 354 } 355 356 /************************************************************************** 357 358 Fill the provided buffer 359 360 Returns: 361 the number of bytes actually read, which will be less than 362 `dst.length` when Eof has been reached and Eof thereafter. 363 364 Params: 365 dst = Where data should be placed. 366 exact = Whether to throw an exception when dst is not 367 filled (an Eof occurs first). Defaults to false. 368 369 **************************************************************************/ 370 371 public final size_t fill (void[] dst, bool exact = false) 372 { 373 size_t len = 0; 374 375 while (len < dst.length) 376 { 377 size_t i = this.read(dst[len .. $]); 378 if (i is Eof) 379 { 380 if (exact && len < dst.length) 381 this.conduit.error("end-of-flow whilst reading"); 382 return (len > 0) ? len : Eof; 383 } 384 len += i; 385 } 386 return len; 387 } 388 389 /*************************************************************************** 390 391 Move the current read location. 392 393 Skips ahead by the specified number of bytes, streaming from 394 the associated conduit as necessary. 395 396 Can also reverse the read position by 'size' bytes, when size 397 is negative. This may be used to support lookahead operations. 398 Note that a negative size will fail where there is not sufficient 399 content available in the buffer (can't _skip beyond the beginning). 400 401 Params: 402 size = The number of bytes to move. 403 404 Returns: 405 `true` if successful, `false` otherwise. 406 407 ***************************************************************************/ 408 409 public final bool skip (int size) 410 { 411 if (size < 0) 412 { 413 size = -size; 414 if (this.index >= size) 415 { 416 this.index -= size; 417 return true; 418 } 419 return false; 420 } 421 return this.slice(size) !is null; 422 } 423 424 /*************************************************************************** 425 426 Move the current read location. 427 428 ***************************************************************************/ 429 430 public final override long seek (long offset, Anchor start = Anchor.Begin) 431 { 432 if (start is Anchor.Current) 433 { 434 // handle this specially because we know this is 435 // buffered - we should take into account the buffer 436 // position when seeking 437 offset -= this.readable; 438 auto bpos = offset + this.limit; 439 440 if (bpos >= 0 && bpos < this.limit) 441 { 442 // the new position is within the current 443 // buffer, skip to that position. 444 this.skip(cast(int) bpos - cast(int) position); 445 446 // see if we can return a valid offset 447 auto pos = this.source.seek(0, Anchor.Current); 448 if (pos != Eof) 449 return pos - this.readable(); 450 return Eof; 451 } 452 // else, position is outside the buffer. Do a real 453 // seek using the adjusted position. 454 } 455 456 this.clear(); 457 return this.source.seek(offset, start); 458 } 459 460 /*************************************************************************** 461 462 Iterator support. 463 464 Upon success, the delegate should return the byte-based index of 465 the consumed pattern (tail end of it). 466 Failure to match a pattern should be indicated by returning an Eof 467 468 Each pattern is expected to be stripped of the delimiter. 469 An end-of-file condition causes trailing content to be 470 placed into the token. Requests made beyond Eof result 471 in empty matches (length is zero). 472 473 Additional iterator and/or reader instances 474 will operate in lockstep when bound to a common buffer. 475 476 Params: 477 scan = The delegate to invoke with the current content. 478 479 Returns: 480 `true` if a token was isolated, `false` otherwise. 481 482 ***************************************************************************/ 483 484 public final bool next (scope size_t delegate (const(void)[]) scan) 485 { 486 while (this.reader(scan) is Eof) 487 { 488 // did we start at the beginning? 489 if (this.position) 490 // yep - move partial token to start of buffer 491 this.compress(); 492 // no more space in the buffer? 493 else if (this.writable is 0) 494 this.extend(); 495 496 verify(this.writable() > 0); 497 498 // read another chunk of data 499 if (this.writer(&this.source.read) is Eof) 500 return false; 501 } 502 return true; 503 } 504 505 /*************************************************************************** 506 507 Reserve the specified space within the buffer, compressing 508 existing content as necessary to make room. 509 510 Returns: 511 the current read point, after compression if that was required. 512 513 ***************************************************************************/ 514 515 public final size_t reserve (size_t space) 516 { 517 verify(space < this.data.length); 518 519 if ((this.data.length - this.index) < space) 520 this.compress(); 521 return this.index; 522 } 523 524 /*************************************************************************** 525 526 Compress buffer space. 527 528 Limit is set to the amount of data remaining. 529 Position is always reset to zero. 530 531 If we have some data left after an export, move it to the front of 532 the buffer and set position to be just after the remains. 533 This is for supporting certain conduits which choose to write just 534 the initial portion of a request. 535 536 Returns: 537 The buffer instance. 538 539 ***************************************************************************/ 540 541 public final BufferedInput compress () 542 { 543 auto r = this.readable(); 544 545 if (this.index > 0 && r > 0) 546 // content may overlap ... 547 memmove(&data[0], &data[this.index], r); 548 549 this.index = 0; 550 this.extent = r; 551 return this; 552 } 553 554 /*************************************************************************** 555 556 Drain buffer content to the specific conduit. 557 558 Returns: 559 the number of bytes written, or Eof. 560 561 Note: 562 Write as much of the buffer that the associated conduit can consume. 563 The conduit is not obliged to consume all content, 564 so some may remain within the buffer. 565 566 ***************************************************************************/ 567 568 public final size_t drain (OutputStream dst) 569 { 570 verify(dst !is null); 571 572 size_t ret = this.reader(&dst.write); 573 this.compress(); 574 return ret; 575 } 576 577 /*************************************************************************** 578 579 Access buffer limit. 580 581 Each buffer has a capacity, a limit, and a position. 582 The capacity is the maximum content a buffer can contain, 583 limit represents the extent of valid content, and position marks 584 the current read location. 585 586 Returns: 587 the limit of readable content within this buffer. 588 589 ***************************************************************************/ 590 591 public final size_t limit () 592 { 593 return this.extent; 594 } 595 596 /*************************************************************************** 597 598 Access buffer capacity. 599 600 Each buffer has a capacity, a limit, and a position. 601 The capacity is the maximum content a buffer can contain, limit 602 represents the extent of valid content, and position marks 603 the current read location. 604 605 Returns: 606 the maximum capacity of this buffer. 607 608 ***************************************************************************/ 609 610 public final size_t capacity () 611 { 612 return this.data.length; 613 } 614 615 /*************************************************************************** 616 617 Access buffer read position. 618 619 Each buffer has a capacity, a limit, and a position. 620 The capacity is the maximum content a buffer can contain, limit 621 represents the extent of valid content, and position marks 622 the current read location. 623 624 Returns: 625 the current read-position within this buffer. 626 627 ***************************************************************************/ 628 629 final size_t position () 630 { 631 return this.index; 632 } 633 634 /*************************************************************************** 635 636 Available content. 637 638 Returns: 639 count of _readable bytes remaining in buffer. 640 This is calculated simply as `this.limit() - this.position()`. 641 642 ***************************************************************************/ 643 644 public final size_t readable () 645 { 646 return this.extent - this.index; 647 } 648 649 /*************************************************************************** 650 651 Cast to a target type without invoking the wrath of the 652 runtime checks for misalignment. Instead, we truncate the 653 array length. 654 655 ***************************************************************************/ 656 657 static inout(T)[] convert (T) (inout(void)[] x) 658 { 659 return (cast(inout(T)*) x.ptr) [0 .. (x.length / T.sizeof)]; 660 } 661 662 /*************************************************************************** 663 664 Clear buffer content. 665 666 Note: 667 Reset 'position' and 'limit' to zero. This effectively 668 clears all content from the buffer. 669 670 ***************************************************************************/ 671 672 public final override BufferedInput flush () 673 { 674 this.index = this.extent = 0; 675 676 // clear the filter chain also 677 if (this.source) 678 super.flush(); 679 return this; 680 } 681 682 /*************************************************************************** 683 684 Set the input stream. 685 686 ***************************************************************************/ 687 688 public final void input (InputStream source) 689 { 690 this.source = source; 691 } 692 693 /*************************************************************************** 694 695 Load the bits from a stream, up to an indicated length, and 696 return them all in an array. 697 698 The function may consume more than the indicated size where additional 699 data is available during a block read operation, but will not wait for 700 more than specified. 701 An Eof terminates the operation. 702 703 Returns: 704 an array representing the content 705 706 Throws: 707 `IOException` on error. 708 709 ***************************************************************************/ 710 711 public final override void[] load (size_t max = size_t.max) 712 { 713 this.load(super.input, super.conduit.bufferSize, max); 714 return this.slice; 715 } 716 717 /*************************************************************************** 718 719 Import content from the specified conduit, expanding as necessary 720 up to the indicated maximum or until an Eof occurs. 721 722 Returns: 723 the number of bytes contained. 724 725 ***************************************************************************/ 726 727 private size_t load (InputStream src, size_t increment, size_t max) 728 { 729 size_t len, count; 730 731 // make some room 732 this.compress(); 733 734 // explicitly resize? 735 if (max != max.max) 736 if ((len = this.writable()) < max) 737 increment = max - len; 738 739 while (count < max) 740 { 741 if (!this.writable()) 742 this.data.length = (this.data.length + increment); 743 if ((len = this.writer(&src.read)) is Eof) 744 break; 745 else 746 count += len; 747 } 748 return count; 749 } 750 751 /*************************************************************************** 752 753 Reset the buffer content. 754 755 Set the backing array with some content readable. 756 Writing to this will either flush it to an associated conduit, 757 or raise an Eof condition. 758 Use clear() to reset the content (make it all writable). 759 760 Params: 761 data = The backing array to buffer within. 762 readable = The number of bytes within data considered valid. 763 764 Returns: 765 The buffer instance. 766 767 ***************************************************************************/ 768 769 private final BufferedInput set (void[] data, size_t readable) 770 { 771 this.data = data; 772 this.extent = readable; 773 774 // reset to start of input 775 this.index = 0; 776 777 return this; 778 } 779 780 /*************************************************************************** 781 782 Available space. 783 784 Returns: 785 count of _writable bytes available in buffer. 786 This is calculated simply as `this.capacity() - this.limit()`. 787 788 ***************************************************************************/ 789 790 private final size_t writable () 791 { 792 return this.data.length - this.extent; 793 } 794 795 /*************************************************************************** 796 797 Extend the buffer by half of its size 798 799 ***************************************************************************/ 800 801 private void extend () 802 { 803 this.data.length = this.data.length + (this.data.length / 2); 804 } 805 } 806 807 808 /******************************************************************************* 809 810 Buffers the flow of data from a upstream output. 811 812 A downstream neighbour can locate and use this buffer instead of creating 813 another instance of their own. 814 815 Don't forget to flush() buffered content before closing. 816 817 Note: 818 upstream is closer to the source, and downstream is further away 819 820 *******************************************************************************/ 821 822 public class BufferedOutput : OutputFilter, OutputBuffer 823 { 824 /// access the sink 825 alias OutputFilter.output output; 826 827 private void[] data; // the raw data buffer 828 private size_t index; // current read position 829 private size_t extent; // limit of valid content 830 private size_t dimension; // maximum extent of content 831 832 /// Notifier that will be called on flush. 833 private void delegate() flush_notifier; 834 835 invariant () 836 { 837 assert (this.index <= this.extent); 838 assert (this.extent <= this.dimension); 839 } 840 841 /*************************************************************************** 842 843 Construct a Buffer upon the provided input stream. 844 845 Params: 846 stream = An input stream. 847 flush_notifier = user specified delegate called after the content 848 of the buffer has been flushed to upstream output. 849 850 ***************************************************************************/ 851 852 public this (OutputStream stream, scope void delegate() flush_notifier = null) 853 { 854 verify(stream !is null); 855 this(stream, stream.conduit.bufferSize, flush_notifier); 856 } 857 858 /*************************************************************************** 859 860 Construct a Buffer upon the provided input stream. 861 862 Params: 863 stream = An input stream. 864 capacity = Desired buffer capacity. 865 flush_notifier = user specified delegate called after the content 866 of the buffer has been flushed to upstream output. 867 868 ***************************************************************************/ 869 870 public this (OutputStream stream, size_t capacity, 871 scope void delegate() flush_notifier = null) 872 { 873 this.set(new ubyte[capacity], 0); 874 this.flush_notifier = flush_notifier; 875 super(this.sink = stream); 876 } 877 878 /*************************************************************************** 879 880 Attempts to share an upstream BufferedOutput, and creates a new 881 instance where there's not a shared one available. 882 883 Where an upstream instance is visible it will be returned. 884 Otherwise, a new instance is created based upon the bufferSize 885 exposed by the associated conduit 886 887 Params: 888 stream = An output stream. 889 890 ***************************************************************************/ 891 892 public static OutputBuffer create (OutputStream stream) 893 { 894 auto sink = stream; 895 auto conduit = sink.conduit; 896 while (cast(Mutator) sink is null) 897 { 898 auto b = cast(OutputBuffer) sink; 899 if (b) 900 return b; 901 if (sink is conduit) 902 break; 903 sink = sink.output; 904 verify(sink !is null); 905 } 906 907 return new BufferedOutput(stream, conduit.bufferSize); 908 } 909 910 /*************************************************************************** 911 912 Retrieve the valid content. 913 914 Returns: 915 A void[] slice of the buffer. 916 917 Returns: 918 a slice of the buffer, from the current position up to the limit 919 of valid content. 920 The content remains in the buffer for future extraction. 921 922 ***************************************************************************/ 923 924 public final void[] slice () 925 { 926 return this.data[this.index .. this.extent]; 927 } 928 929 /*************************************************************************** 930 931 Emulate OutputStream.write(). 932 933 Appends src content to the buffer, flushing to an attached conduit 934 as necessary. An IOException is thrown upon write failure. 935 936 Params: 937 src = The content to write. 938 939 Returns: 940 the number of bytes written, which may be less than provided 941 (conceptually). 942 943 ***************************************************************************/ 944 945 public final override size_t write (const(void)[] src) 946 { 947 this.append(src.ptr, src.length); 948 return src.length; 949 } 950 951 /*************************************************************************** 952 953 Append content. 954 955 Append an array to this buffer, flush to the conduit as necessary. 956 This is often used in lieu of a Writer. 957 958 Params: 959 src = The content to _append. 960 961 Returns: 962 a chaining reference if all content was written. 963 964 Throws: 965 an IOException indicating Eof or Eob if not. 966 967 ***************************************************************************/ 968 969 public final BufferedOutput append (const(void)[] src) 970 { 971 return this.append(src.ptr, src.length); 972 } 973 974 /*************************************************************************** 975 976 Append content. 977 978 Append an array to this buffer, flush to the conduit as necessary. 979 This is often used in lieu of a Writer. 980 981 Params: 982 src = The content to _append. 983 length = The number of bytes in src. 984 985 Returns: 986 a chaining reference if all content was written. 987 988 Throws: 989 an IOException indicating Eof or Eob if not. 990 991 ***************************************************************************/ 992 993 public final BufferedOutput append (const(void)* src, size_t length) 994 { 995 if (length > this.writable) 996 { 997 this.flush(); 998 999 // check for pathological case 1000 if (length > this.dimension) 1001 do { 1002 auto written = this.sink.write(src [0 .. length]); 1003 if (written is Eof) 1004 this.conduit.error("end-of-flow whilst writing"); 1005 length -= written; 1006 src += written; 1007 } while (length > this.dimension); 1008 } 1009 1010 // avoid "out of bounds" test on zero length 1011 if (length) 1012 { 1013 // content may overlap ... 1014 memmove(&this.data[this.extent], src, length); 1015 this.extent += length; 1016 } 1017 return this; 1018 } 1019 1020 /*************************************************************************** 1021 1022 Available space. 1023 1024 Returns: 1025 count of _writable bytes available in buffer. 1026 This is calculated as `capacity() - limit()`. 1027 1028 ***************************************************************************/ 1029 1030 public final size_t writable () 1031 { 1032 return this.dimension - this.extent; 1033 } 1034 1035 /*************************************************************************** 1036 1037 Access buffer limit. 1038 1039 Each buffer has a capacity, a limit, and a position. 1040 The capacity is the maximum content a buffer can contain, 1041 limit represents the extent of valid content, and position marks 1042 the current read location. 1043 1044 Returns: 1045 the limit of readable content within this buffer. 1046 1047 ***************************************************************************/ 1048 1049 public final size_t limit () 1050 { 1051 return this.extent; 1052 } 1053 1054 /*************************************************************************** 1055 1056 Access buffer capacity. 1057 1058 Each buffer has a capacity, a limit, and a position. 1059 The capacity is the maximum content a buffer can contain, 1060 limit represents the extent of valid content, and position marks 1061 the current read location. 1062 1063 Returns: 1064 the maximum capacity of this buffer. 1065 1066 ***************************************************************************/ 1067 1068 public final size_t capacity () 1069 { 1070 return this.dimension; 1071 } 1072 1073 /*************************************************************************** 1074 1075 Truncate the buffer within its extent. 1076 1077 Returns: 1078 `true` if the new length is valid, `false` otherwise. 1079 1080 ***************************************************************************/ 1081 1082 public final bool truncate (size_t length) 1083 { 1084 if (length <= this.data.length) 1085 { 1086 this.extent = length; 1087 return true; 1088 } 1089 return false; 1090 } 1091 1092 /*************************************************************************** 1093 1094 Cast to a target type without invoking the wrath of the 1095 runtime checks for misalignment. Instead, we truncate the 1096 array length. 1097 1098 ***************************************************************************/ 1099 1100 static T[] convert(T)(void[] x) 1101 { 1102 return (cast(T*) x.ptr) [0 .. (x.length / T.sizeof)]; 1103 } 1104 1105 /*************************************************************************** 1106 1107 Flush all buffer content to the specific conduit. 1108 1109 Flush the contents of this buffer. 1110 This will block until all content is actually flushed via the associated 1111 conduit, whereas `drain()` will not. 1112 1113 Throws: 1114 an IOException on premature Eof. 1115 1116 ***************************************************************************/ 1117 1118 final override BufferedOutput flush () 1119 { 1120 while (this.readable() > 0) 1121 { 1122 auto ret = this.reader(&this.sink.write); 1123 if (ret is Eof) 1124 this.conduit.error("end-of-flow whilst writing"); 1125 } 1126 1127 // flush the filter chain also 1128 this.clear(); 1129 super.flush; 1130 1131 if (this.flush_notifier) 1132 { 1133 this.flush_notifier(); 1134 } 1135 1136 return this; 1137 } 1138 1139 /*************************************************************************** 1140 1141 Copy content via this buffer from the provided src conduit. 1142 1143 The src conduit has its content transferred through this buffer via 1144 a series of fill & drain operations, 1145 until there is no more content available. 1146 The buffer content should be explicitly flushed by the caller. 1147 1148 Throws: 1149 an IOException on premature Eof. 1150 1151 ***************************************************************************/ 1152 1153 final override BufferedOutput copy (InputStream src, size_t max = -1) 1154 { 1155 size_t chunk, copied; 1156 1157 while (copied < max && (chunk = this.writer(&src.read)) != Eof) 1158 { 1159 copied += chunk; 1160 1161 // don't drain until we actually need to 1162 if (this.writable is 0) 1163 if (this.drain(this.sink) is Eof) 1164 this.conduit.error("end-of-flow whilst writing"); 1165 } 1166 return this; 1167 } 1168 1169 /*************************************************************************** 1170 1171 Flushes the buffer and closes the stream. 1172 1173 ***************************************************************************/ 1174 1175 final override void close ( ) 1176 { 1177 this.flush(); 1178 super.close(); 1179 } 1180 1181 /*************************************************************************** 1182 1183 Drain buffer content to the specific conduit. 1184 1185 Write as much of the buffer that the associated conduit can consume. 1186 The conduit is not obliged to consume all content, 1187 so some may remain within the buffer. 1188 1189 Returns: 1190 the number of bytes written, or Eof. 1191 1192 ***************************************************************************/ 1193 1194 final size_t drain (OutputStream dst) 1195 { 1196 verify(dst !is null); 1197 1198 size_t ret = this.reader(&dst.write); 1199 this.compress(); 1200 return ret; 1201 } 1202 1203 /*************************************************************************** 1204 1205 Clear buffer content. 1206 1207 Reset 'position' and 'limit' to zero. 1208 This effectively clears all content from the buffer. 1209 1210 ***************************************************************************/ 1211 1212 final BufferedOutput clear () 1213 { 1214 this.index = this.extent = 0; 1215 return this; 1216 } 1217 1218 /*************************************************************************** 1219 1220 Set the output stream. 1221 1222 ***************************************************************************/ 1223 1224 final void output (OutputStream sink) 1225 { 1226 this.sink = sink; 1227 } 1228 1229 /*************************************************************************** 1230 1231 Seek within this stream 1232 1233 Any and all buffered output is disposed before the upstream is invoked. 1234 Use an explicit `flush()` to emit content prior to seeking. 1235 1236 ***************************************************************************/ 1237 1238 final override long seek (long offset, Anchor start = Anchor.Begin) 1239 { 1240 this.clear(); 1241 return super.seek(offset, start); 1242 } 1243 1244 /*************************************************************************** 1245 1246 Write into this buffer. 1247 1248 Exposes the raw data buffer at the current _write position, 1249 The delegate is provided with a void[] representing space 1250 available within the buffer at the current _write position. 1251 1252 The delegate should return the appropriate number of bytes 1253 if it writes valid content, or Eof on error. 1254 1255 Params: 1256 dg = The callback to provide buffer access to. 1257 1258 Returns: 1259 the delegate return's value 1260 1261 ***************************************************************************/ 1262 1263 final size_t writer (scope size_t delegate (void[]) dg) 1264 { 1265 auto count = dg (this.data[this.extent..this.dimension]); 1266 1267 if (count != Eof) 1268 { 1269 this.extent += count; 1270 verify(this.extent <= this.dimension); 1271 } 1272 return count; 1273 } 1274 1275 /*************************************************************************** 1276 1277 Read directly from this buffer. 1278 1279 Exposes the raw data buffer at the current _read position. 1280 The delegate is provided with a void[] representing the available data, 1281 and should return zero to leave the current _read position intact. 1282 1283 If the delegate consumes data, it should return the number of 1284 bytes consumed; or Eof to indicate an error. 1285 1286 Params: 1287 dg = Callback to provide buffer access to. 1288 1289 Returns: 1290 the delegate's return value. 1291 1292 ***************************************************************************/ 1293 1294 private final size_t reader (scope size_t delegate (const(void)[]) dg) 1295 { 1296 auto count = dg (this.data[this.index..this.extent]); 1297 1298 if (count != Eof) 1299 { 1300 this.index += count; 1301 verify(this.index <= this.extent); 1302 } 1303 return count; 1304 } 1305 1306 /*************************************************************************** 1307 1308 Available content. 1309 1310 Returns: 1311 count of _readable bytes remaining in buffer. 1312 This is calculated simply as `limit() - position()`. 1313 1314 ***************************************************************************/ 1315 1316 private final size_t readable () 1317 { 1318 return this.extent - this.index; 1319 } 1320 1321 /*************************************************************************** 1322 1323 Reset the buffer content. 1324 1325 Set the backing array with some content readable. 1326 Writing to this will either flush it to an associated conduit, 1327 or raise an Eof condition. 1328 Use clear() to reset the content (make it all writable). 1329 1330 Params: 1331 data = The backing array to buffer within. 1332 readable = The number of bytes within data considered valid. 1333 1334 Returns: 1335 The buffer instance. 1336 1337 ***************************************************************************/ 1338 1339 private final BufferedOutput set (void[] data, size_t readable) 1340 { 1341 this.data = data; 1342 this.extent = readable; 1343 this.dimension = data.length; 1344 1345 // reset to start of input 1346 this.index = 0; 1347 1348 return this; 1349 } 1350 1351 /*************************************************************************** 1352 1353 Compress buffer space. 1354 1355 Limit is set to the amount of data remaining. 1356 Position is always reset to zero. 1357 1358 If we have some data left after an export, move it to front of the 1359 buffer and set position to be just after the remains. 1360 This is for supporting certain conduits which choose to write just 1361 the initial portion of a request. 1362 1363 Returns: 1364 The buffer instance. 1365 1366 ***************************************************************************/ 1367 1368 private final BufferedOutput compress () 1369 { 1370 size_t r = this.readable(); 1371 1372 if (this.index > 0 && r > 0) 1373 // content may overlap ... 1374 memmove(&data[0], &data[this.index], r); 1375 1376 this.index = 0; 1377 this.extent = r; 1378 return this; 1379 } 1380 } 1381 1382 1383 unittest 1384 { 1385 scope device = new MemoryDevice; 1386 scope buffer = new BufferedInput(device, 16); 1387 1388 device.write( 1389 "En 1815, M. Charles-François-Bienvenu Myriel était évêque de Digne. " ~ 1390 "C’était un vieillard d’environ soixante-quinze ans; " ~ 1391 "il occupait le siège de Digne depuis 1806."); 1392 device.seek(0); 1393 1394 size_t finder (const(void)[] raw) 1395 { 1396 auto text = cast(cstring) raw; 1397 if (raw.length >= 5 && raw[$ - 5 .. $] == "1806.") 1398 return raw.length - 5; 1399 return BufferedInput.Eof; 1400 } 1401 1402 test(buffer.next(&finder)); 1403 }