aboutsummaryrefslogtreecommitdiffstats
path: root/include/jau/byte_stream.hpp
blob: 11797c352221156c63a48124f704f8ec178d62d0 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
/*
 * Author: Sven Gothel <sgothel@jausoft.com>
 * Copyright (c) 2021 Gothel Software e.K.
 *
 * Permission is hereby granted, free of charge, to any person obtaining
 * a copy of this software and associated documentation files (the
 * "Software"), to deal in the Software without restriction, including
 * without limitation the rights to use, copy, modify, merge, publish,
 * distribute, sublicense, and/or sell copies of the Software, and to
 * permit persons to whom the Software is furnished to do so, subject to
 * the following conditions:
 *
 * The above copyright notice and this permission notice shall be
 * included in all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
 * LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
 * OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
 * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */

#ifndef JAU_BYTE_STREAM_HPP_
#define JAU_BYTE_STREAM_HPP_

#include <string>
#include <cstdint>
#include <functional>
#include <thread>

#include <jau/basic_types.hpp>
#include <jau/ringbuffer.hpp>
#include <jau/file_util.hpp>

#include <jau/io_util.hpp>

using namespace jau::fractions_i64_literals;

namespace jau::io {

    /** \addtogroup IOUtils
     *
     *  @{
     */

    /**
     * Mimic std::ios_base::iostate for state functionality, see iostate_func.
     *
     * This `enum class` type fulfills `C++ named requirements: BitmaskType`.
     */
    enum class iostate : uint32_t {
      /** No error occurred nor has EOS being reached. Value is no bit set! */
      none = 0,

      /** No error occurred nor has EOS being reached. Value is no bit set! */
      goodbit = 0,

      /** Irrecoverable stream error, including loss of integrity of the underlying stream or media. */
      badbit  = 1 << 0,

      /** An input operation reached the end of its stream. */
      eofbit  = 1 << 1,

      /** Input or output operation failed (formatting or extraction error). */
      failbit = 1 << 2
    };
    constexpr uint32_t number(const iostate rhs) noexcept {
        return static_cast<uint32_t>(rhs);
    }
    constexpr iostate operator ~(const iostate rhs) noexcept {
        return static_cast<iostate> ( ~number(rhs) );
    }
    constexpr iostate operator ^(const iostate lhs, const iostate rhs) noexcept {
        return static_cast<iostate> ( number(lhs) ^ number(rhs) );
    }
    constexpr iostate operator |(const iostate lhs, const iostate rhs) noexcept {
        return static_cast<iostate> ( number(lhs) | number(rhs) );
    }
    constexpr iostate operator &(const iostate lhs, const iostate rhs) noexcept {
        return static_cast<iostate> ( number(lhs) & number(rhs) );
    }
    constexpr iostate& operator |=(iostate& lhs, const iostate rhs) noexcept {
        lhs = static_cast<iostate> ( number(lhs) | number(rhs) );
        return lhs;
    }
    constexpr iostate& operator &=(iostate& lhs, const iostate rhs) noexcept {
        lhs = static_cast<iostate> ( number(lhs) & number(rhs) );
        return lhs;
    }
    constexpr iostate& operator ^=(iostate& lhs, const iostate rhs) noexcept {
        lhs = static_cast<iostate> ( number(lhs) ^ number(rhs) );
        return lhs;
    }
    constexpr bool operator ==(const iostate lhs, const iostate rhs) noexcept {
        return number(lhs) == number(rhs);
    }
    constexpr bool operator !=(const iostate lhs, const iostate rhs) noexcept {
        return !( lhs == rhs );
    }
    std::string to_string(const iostate mask) noexcept;

    /**
     * Supporting std::basic_ios's iostate functionality for all ByteInStream implementations.
     */
    class iostate_func {
        private:
            iostate m_state;

        protected:
            constexpr iostate rdstate_impl() const noexcept { return m_state; }
            constexpr void setstate_impl(iostate state) const noexcept { const_cast<iostate_func*>(this)->m_state |= state; }

        public:
            iostate_func() noexcept
            : m_state( iostate::goodbit ) {}

            iostate_func(const iostate_func& o) noexcept = default;
            iostate_func(iostate_func&& o) noexcept = default;
            iostate_func& operator=(const iostate_func &o) noexcept = default;
            iostate_func& operator=(iostate_func &&o) noexcept = default;

            virtual ~iostate_func() noexcept {}

            /** Clears state flags by assignment to the given value. */
            virtual void clear(const iostate state = iostate::goodbit) noexcept { m_state = state; }

            /**
             * Returns the current state flags.
             *
             * Method is marked `virtual` to allow implementations with asynchronous resources
             * to determine or update the current iostate.
             *
             * Method is used throughout all query members and setstate(),
             * hence they all will use the updated state from a potential override implementation.
             */
            virtual iostate rdstate() const noexcept { return m_state; }

            /** Sets state flags, by keeping its previous bits. */
            void setstate(const iostate state) noexcept { clear( rdstate() | state ); }

            /** Checks if no error nor eof() has occurred i.e. I/O operations are available. */
            bool good() const noexcept
            { return iostate::goodbit == rdstate(); }

            /** Checks if end-of-file has been reached. */
            bool eof() const noexcept
            { return iostate::none != ( rdstate() & iostate::eofbit ); }

            /** Checks if an error has occurred. */
            bool fail() const noexcept
            { return iostate::none != ( rdstate() & ( iostate::badbit | iostate::failbit ) ); }

            /** Checks if an error has occurred, synonym of fail(). */
            bool operator!() const noexcept { return fail(); }

            /** Checks if no error has occurred, synonym of !fail(). */
            explicit operator bool() const noexcept { return !fail(); }

            /** Checks if a non-recoverable error has occurred. */
            bool bad() const noexcept
            { return iostate::none != ( rdstate() & iostate::badbit ); }
    };

    /**
     * Abstract byte input stream object.
     *
     * @anchor byte_in_stream_properties
     * ### ByteInStream Properties
     * The byte input stream can originate from a local source w/o delay,
     * remote URL like http connection or even from another thread feeding the input buffer.<br />
     * Both latter asynchronous resources may expose blocking properties
     * in available().
     *
     * Asynchronous resources benefit from knowing their content size,
     * as their available() implementation may avoid
     * blocking and waiting for requested bytes available
     * if the stream is already beyond its scope.
     *
     * All method implementations are of `noexcept`.
     *
     * One may use fail() to detect whether an error has occurred,
     * while end_of_data() not only covers the end-of-stream (EOS) case but includes fail().
     *
     * @see @ref byte_in_stream_properties "ByteInStream Properties"
     */
    class ByteInStream : public iostate_func
    {
        public:
            ByteInStream() noexcept
            : iostate_func() {}

            virtual ~ByteInStream() noexcept = default;
            ByteInStream& operator=(const ByteInStream&) = delete;
            ByteInStream(const ByteInStream&) = delete;

            /** Checks if the stream has an associated file. */
            virtual bool is_open() const noexcept = 0;

            /**
             * Close the stream if supported by the underlying mechanism.
             */
            virtual void close() noexcept = 0;

            /**
             * Return whether n bytes are available in the input stream,
             * if has_content_size() or using an asynchronous source.
             *
             * If !has_content_size() and not being an asynchronous source,
             * !end_of_data() is returned.
             *
             * Method may be blocking when using an asynchronous source
             * up until the requested bytes are available.
             *
             * A subsequent call to read() shall return immediately with at least
             * the requested numbers of bytes available,
             * if has_content_size() or using an asynchronous source.
             *
             * See details of the implementing class.
             *
             * @param n byte count to wait for
             * @return true if n bytes are available, otherwise false
             *
             * @see has_content_size()
             * @see read()
             * @see @ref byte_in_stream_properties "ByteInStream Properties"
             */
            virtual bool available(size_t n) noexcept  = 0;

            /**
             * Read from the source. Moves the internal offset so that every
             * call to read will return a new portion of the source.
             *
             * Use available() to try to wait for a certain amount of bytes available.
             *
             * This method shall only block until `min(available, length)` bytes are transfered.
             *
             * See details of the implementing class.
             *
             * @param out the byte array to write the result to
             * @param length the length of the byte array out
             * @return length in bytes that was actually read and put into out
             *
             * @see available()
             * @see @ref byte_in_stream_properties "ByteInStream Properties"
             */
            [[nodiscard]] virtual size_t read(void* out, size_t length) noexcept = 0;

            /**
             * Read one byte.
             * @param out the byte to read to
             * @return length in bytes that was actually read and put
             * into out
             */
            size_t read(uint8_t& out) noexcept;

            /**
             * Read from the source but do not modify the internal
             * offset. Consecutive calls to peek() will return portions of
             * the source starting at the same position.
             *
             * @param out the byte array to write the output to
             * @param length the length of the byte array out
             * @param peek_offset the offset into the stream to read at
             * @return length in bytes that was actually read and put into out
             */
            [[nodiscard]] virtual size_t peek(void* out, size_t length, size_t peek_offset) noexcept = 0;

            /**
             * Peek at one byte.
             * @param out an output byte
             * @return length in bytes that was actually read and put
             * into out
             */
            size_t peek(uint8_t& out) noexcept;

            /**
             * Discard the next N bytes of the data
             * @param N the number of bytes to discard
             * @return number of bytes actually discarded
             */
            size_t discard(size_t N) noexcept;

            /**
             * Test whether the source still has data that can be read, synonym for !good().
             *
             * Hence this includes errors in the underlying implementation, see fail()
             *
             * @return true if there is no more data to read, false otherwise
             * @see good()
             * @see fail()
             */
            bool end_of_data() const noexcept  { return !good(); }

            /**
             * return the id of this data source
             * @return std::string representing the id of this data source
             */
            virtual std::string id() const noexcept { return ""; }

            /**
             * Returns the input position indicator, similar to std::basic_istream.
             *
             * @return number of bytes read so far.
             */
            virtual uint64_t tellg() const noexcept = 0;

            /**
             * Returns true if implementation is aware of content_size(), otherwise false.
             * @see content_size()
             */
            virtual bool has_content_size() const noexcept = 0;

            /**
             * Returns the content_size if known.
             * @see has_content_size()
             */
            virtual uint64_t content_size() const noexcept = 0;

            virtual std::string to_string() const noexcept = 0;
    };

    /**
     * Secure Memory-Based byte input stream
     */
    class ByteInStream_SecMemory final : public ByteInStream {
       public:
          size_t read(void*, size_t) noexcept override;

          size_t peek(void*, size_t, size_t) noexcept override;

          bool available(size_t n) noexcept override;

          /**
          * Construct a secure memory source that reads from a string
          * @param in the string to read from
          */
          explicit ByteInStream_SecMemory(const std::string& in);

          /**
          * Construct a secure memory source that reads from a byte array
          * @param in the byte array to read from
          * @param length the length of the byte array
          */
          ByteInStream_SecMemory(const uint8_t in[], size_t length)
          : m_source(in, in + length), m_offset(0) {}

          /**
          * Construct a secure memory source that reads from a secure_vector
          * @param in the MemoryRegion to read from
          */
          explicit ByteInStream_SecMemory(const io::secure_vector<uint8_t>& in)
          : m_source(in), m_offset(0) {}

          /**
          * Construct a secure memory source that reads from a std::vector
          * @param in the MemoryRegion to read from
          */
          explicit ByteInStream_SecMemory(const std::vector<uint8_t>& in)
          : m_source(in.begin(), in.end()), m_offset(0) {}

          bool is_open() const noexcept override { return true; }

          void close() noexcept override;

          ~ByteInStream_SecMemory() noexcept override { close(); }

          uint64_t tellg() const noexcept override { return m_offset; }

          bool has_content_size() const noexcept override { return true; }

          uint64_t content_size() const noexcept override { return m_source.size(); }

          std::string to_string() const noexcept override;

       private:
          io::secure_vector<uint8_t> m_source;
          size_t m_offset;
    };

    /**
     * File based byte input stream, including named file descriptor.
     *
     * Implementation mimics std::ifstream via OS level file descriptor (FD) operations,
     * giving more flexibility, allowing reusing existing FD and enabling openat() operations.
     *
     * If source path denotes a named file descriptor, i.e. jau::fs::file_stats::is_fd() returns true,
     * has_content_size() returns false and available() returns true as long the stream is open and EOS hasn't occurred.
     */
    class ByteInStream_File final : public ByteInStream {
       private:
          jau::fs::file_stats stats;

          int m_fd;

          bool m_has_content_length;
          uint64_t m_content_size;
          uint64_t m_bytes_consumed;
          uint64_t get_available() const noexcept { return m_has_content_length ? m_content_size - m_bytes_consumed : 0; }

       public:
          size_t read(void*, size_t) noexcept override;
          size_t peek(void*, size_t, size_t) noexcept override;
          bool available(size_t n) noexcept override;

          bool is_open() const noexcept override { return 0 <= m_fd; }

          std::string id() const noexcept override { return stats.path(); }

          /**
           * Returns the file descriptor if is_open(), otherwise -1 for no file descriptor.
           *
           * @see is_open()
           */
          int fd() const noexcept { return m_fd; }

          /**
           * Construct a stream based byte input stream from filesystem path
           *
           * In case the given path is a local file URI starting with `file://`, see jau::io::uri::is_local_file_protocol(),
           * the leading `file://` is cut off and the remainder being used.
           *
           * @param path the path to the file, maybe a local file URI
           */
          ByteInStream_File(const std::string& path) noexcept;

          /**
           * Construct a stream based byte input stream from filesystem path and parent directory file descriptor
           *
           * In case the given path is a local file URI starting with `file://`, see jau::io::uri::is_local_file_protocol(),
           * the leading `file://` is cut off and the remainder being used.
           *
           * @param dirfd parent directory file descriptor
           * @param path the path to the file, maybe a local file URI
           */
          ByteInStream_File(const int dirfd, const std::string& path) noexcept;

          /**
           * Construct a stream based byte input stream by duplicating given file descriptor
           *
           * In case the given path is a local file URI starting with `file://`, see jau::io::uri::is_local_file_protocol(),
           * the leading `file://` is cut off and the remainder being used.
           *
           * @param fd file descriptor to duplicate leaving the given `fd` untouched
           */
          ByteInStream_File(const int fd) noexcept;

          ByteInStream_File(const ByteInStream_File&) = delete;

          ByteInStream_File& operator=(const ByteInStream_File&) = delete;

          void close() noexcept override;

          virtual ~ByteInStream_File() noexcept override { close(); }

          uint64_t tellg() const noexcept override { return m_bytes_consumed; }

          bool has_content_size() const noexcept override { return m_has_content_length; }

          uint64_t content_size() const noexcept override { return m_content_size; }

          std::string to_string() const noexcept override;
    };

    /**
     * Ringbuffer-Based byte input stream with a URL connection provisioned data feed.
     *
     * Standard implementation uses [curl](https://curl.se/),
     * hence all [*libcurl* network protocols](https://curl.se/docs/url-syntax.html) are supported,
     * jau::io::uri::supported_protocols().
     */
    class ByteInStream_URL final : public ByteInStream {
        public:
            /**
             * Check whether n bytes are available in the input stream.
             *
             * Wait up to timeout duration given in constructor until n bytes become available, where fractions_i64::zero waits infinitely.
             *
             * This method is blocking.
             *
             * @param n byte count to wait for
             * @return true if n bytes are available, otherwise false
             *
             * @see read()
             * @see @ref byte_in_stream_properties "ByteInStream Properties"
             */
            bool available(size_t n) noexcept override;

            /**
             * Read from the source. Moves the internal offset so that every
             * call to read will return a new portion of the source.
             *
             * Use available() to wait and ensure a certain amount of bytes are available.
             *
             * This method is not blocking beyond the transfer of `min(available, length)` bytes.
             *
             * @param out the byte array to write the result to
             * @param length the length of the byte array out
             * @return length in bytes that was actually read and put into out
             *
             * @see available()
             * @see @ref byte_in_stream_properties "ByteInStream Properties"
             */
            size_t read(void* out, size_t length) noexcept override;

            size_t peek(void* out, size_t length, size_t peek_offset) noexcept override;

            iostate rdstate() const noexcept override;

            std::string id() const noexcept override { return m_url; }

            /**
             * Construct a ringbuffer backed Http byte input stream
             * @param url the URL of the data to read
             * @param timeout maximum duration in fractions of seconds to wait @ available() for next bytes, where fractions_i64::zero waits infinitely
             */
            ByteInStream_URL(const std::string& url, const jau::fraction_i64& timeout) noexcept;

            ByteInStream_URL(const ByteInStream_URL&) = delete;

            ByteInStream_URL& operator=(const ByteInStream_URL&) = delete;

            bool is_open() const noexcept override;

            void close() noexcept override;

            ~ByteInStream_URL() noexcept override { close(); }

            uint64_t tellg() const noexcept override { return m_bytes_consumed; }

            bool has_content_size() const noexcept override { return m_has_content_length; }

            uint64_t content_size() const noexcept override { return m_content_size; }

            std::string to_string() const noexcept override;

        private:
            uint64_t get_available() const noexcept { return m_has_content_length ? m_content_size - m_bytes_consumed : 0; }
            std::string to_string_int() const noexcept;

            const std::string m_url;
            jau::fraction_i64 m_timeout;
            ByteRingbuffer m_buffer;
            jau::relaxed_atomic_bool m_has_content_length;
            jau::relaxed_atomic_uint64 m_content_size;
            jau::relaxed_atomic_uint64 m_total_xfered;
            relaxed_atomic_async_io_result_t m_result;
            std::unique_ptr<std::thread> m_url_thread;
            uint64_t m_bytes_consumed;
    };

    /**
     * Parses the given path_or_uri, if it matches a supported protocol, see jau::io::uri::protocol_supported(),
     * but is not a local file, see jau::io::uri::is_local_file_protocol(), ByteInStream_URL is being attempted.
     *
     * If the above fails, ByteInStream_File is attempted.
     *
     * If non of the above leads to a ByteInStream without ByteInStream::fail(), nullptr is returned.
     *
     * @param path_or_uri given path or uri for with a ByteInStream instance shall be established.
     * @param timeout in case `path_or_uri` resolves to ByteInStream_URL, timeout is being used as maximum duration to wait for next bytes at ByteInStream_URL::available(), defaults to 20_s
     * @return a working ByteInStream w/o ByteInStream::fail() or nullptr
     */
    std::unique_ptr<ByteInStream> to_ByteInStream(const std::string& path_or_uri, jau::fraction_i64 timeout=20_s) noexcept;

    /**
     * Ringbuffer-Based byte input stream with an externally provisioned data feed.
     */
    class ByteInStream_Feed final : public ByteInStream {
        public:
            /**
             * Check whether n bytes are available in the input stream.
             *
             * Wait up to timeout duration given in constructor until n bytes become available, where fractions_i64::zero waits infinitely.
             *
             * This method is blocking.
             *
             * @param n byte count to wait for
             * @return true if n bytes are available, otherwise false
             *
             * @see read()
             * @see @ref byte_in_stream_properties "ByteInStream Properties"
             */
            bool available(size_t n) noexcept override;

            /**
             * Read from the source. Moves the internal offset so that every
             * call to read will return a new portion of the source.
             *
             * Use available() to wait and ensure a certain amount of bytes are available.
             *
             * This method is not blocking beyond the transfer of `min(available, length)` bytes.
             *
             * @param out the byte array to write the result to
             * @param length the length of the byte array out
             * @return length in bytes that was actually read and put into out
             *
             * @see available()
             * @see @ref byte_in_stream_properties "ByteInStream Properties"
             */
            size_t read(void* out, size_t length) noexcept override;

            size_t peek(void* out, size_t length, size_t peek_offset) noexcept override;

            iostate rdstate() const noexcept override;

            std::string id() const noexcept override { return m_id; }

            /**
             * Construct a ringbuffer backed externally provisioned byte input stream
             * @param id_name arbitrary identifier for this instance
             * @param timeout maximum duration in fractions of seconds to wait @ available() and write(), where fractions_i64::zero waits infinitely
             */
            ByteInStream_Feed(const std::string& id_name, const jau::fraction_i64& timeout) noexcept;

            ByteInStream_Feed(const ByteInStream_Feed&) = delete;

            ByteInStream_Feed& operator=(const ByteInStream_Feed&) = delete;

            bool is_open() const noexcept override;

            void close() noexcept override;

            ~ByteInStream_Feed() noexcept override { close(); }

            uint64_t tellg() const noexcept override { return m_bytes_consumed; }

            bool has_content_size() const noexcept override { return m_has_content_length; }

            uint64_t content_size() const noexcept override { return m_content_size; }

            /**
             * Interrupt a potentially blocked reader.
             *
             * Call this method if intended to abort streaming and to interrupt the reader thread's potentially blocked available() call,
             * i.e. done at set_eof()
             *
             * @see set_eof()
             */
            void interruptReader() noexcept {
                m_buffer.interruptReader();
            }

            /**
             * Write given bytes to the async ringbuffer.
             *
             * Wait up to timeout duration given in constructor until ringbuffer space is available, where fractions_i64::zero waits infinitely.
             *
             * This method is blocking.
             *
             * @param n byte count to wait for
             * @param in the byte array to transfer to the async ringbuffer
             * @param length the length of the byte array in
             */
            void write(uint8_t in[], size_t length) noexcept;

            /**
             * Set known content size, informal only.
             * @param content_length the content size in bytes
             */
            void set_content_size(const uint64_t size) noexcept {
                m_content_size = size;
                m_has_content_length = true;
            }

            /**
             * Set end-of-data (EOS), i.e. when feeder completed provisioning bytes.
             *
             * Implementation issues interruptReader() to unblock a potentially blocked reader thread.
             *
             * @param result should be either result_t::FAILED or result_t::SUCCESS.
             *
             * @see interruptReader()
             */
            void set_eof(const async_io_result_t result) noexcept;

            std::string to_string() const noexcept override;

        private:
            uint64_t get_available() const noexcept { return m_has_content_length ? m_content_size - m_bytes_consumed : 0; }
            std::string to_string_int() const noexcept;

            const std::string m_id;
            jau::fraction_i64 m_timeout;
            ByteRingbuffer m_buffer;
            jau::relaxed_atomic_bool m_has_content_length;
            jau::relaxed_atomic_uint64 m_content_size;
            jau::relaxed_atomic_uint64 m_total_xfered;
            relaxed_atomic_async_io_result_t m_result;
            uint64_t m_bytes_consumed;
    };

    /**
     * Wrapped byte input stream with the capability
     * to record the read byte stream at will.
     *
     * Peek'ed bytes won't be recorded, only read bytes.
     */
    class ByteInStream_Recorder final : public ByteInStream {
        public:
            size_t read(void*, size_t) noexcept override;

            size_t peek(void* out, size_t length, size_t peek_offset) noexcept override {
                return m_parent.peek(out, length, peek_offset);
            }

            bool available(size_t n) noexcept override {
                return m_parent.available(n);
            }

            void clear(iostate state = iostate::goodbit) noexcept override { m_parent.clear( state ); }
            iostate rdstate() const noexcept override { return m_parent.rdstate(); }

            std::string id() const noexcept override { return m_parent.id(); }

            /**
             * Construct a byte input stream wrapper using the given parent ByteInStream.
             * @param parent the parent ByteInStream
             * @param buffer a user defined buffer for the recording
             */
            ByteInStream_Recorder(ByteInStream& parent, io::secure_vector<uint8_t>& buffer) noexcept
            : m_parent(parent), m_bytes_consumed(0), m_buffer(buffer), m_rec_offset(0), m_is_recording(false) {};

            ByteInStream_Recorder(const ByteInStream_Recorder&) = delete;

            ByteInStream_Recorder& operator=(const ByteInStream_Recorder&) = delete;

            bool is_open() const noexcept override { return m_parent.is_open(); }

            void close() noexcept override;

            ~ByteInStream_Recorder() noexcept override { close(); }

            uint64_t tellg() const noexcept override { return m_bytes_consumed; }

            bool has_content_size() const noexcept override { return m_parent.has_content_size(); }

            uint64_t content_size() const noexcept override { return m_parent.content_size(); }

            /**
             * Starts the recording.
             * <p>
             * A potential previous recording will be cleared.
             * </p>
             */
            void start_recording() noexcept;

            /**
             * Stops the recording.
             * <p>
             * The recording persists.
             * </p>
             */
            void stop_recording() noexcept;

            /**
             * Clears the recording.
             * <p>
             * If the recording was ongoing, also stops the recording.
             * </p>
             */
            void clear_recording() noexcept;

            /** Returns the reference of the recording buffer given by user. */
            io::secure_vector<uint8_t>& get_recording() noexcept { return m_buffer; }

            size_t get_bytes_recorded() noexcept { return m_buffer.size(); }

            /** Returns the recording start position. */
            uint64_t get_recording_start_pos() noexcept { return m_rec_offset; }

            bool is_recording() noexcept { return m_is_recording; }

            std::string to_string() const noexcept override;

        private:
            ByteInStream& m_parent;
            uint64_t m_bytes_consumed;
            io::secure_vector<uint8_t>& m_buffer;
            uint64_t m_rec_offset;
            bool m_is_recording;
    };

    /**
     * Abstract byte output stream object,
     * to write data to a sink.
     *
     * All method implementations are of `noexcept`.
     *
     * One may use fail() to detect whether an error has occurred.
     */
    class ByteOutStream : public iostate_func
    {
        public:
            ByteOutStream() = default;
            virtual ~ByteOutStream() noexcept = default;
            ByteOutStream& operator=(const ByteOutStream&) = delete;
            ByteOutStream(const ByteOutStream&) = delete;

            /** Checks if the stream has an associated file. */
            virtual bool is_open() const noexcept = 0;

            /**
             * Close the stream if supported by the underlying mechanism.
             */
            virtual void close() noexcept = 0;

            /**
             * Write to the data sink. Moves the internal offset so that every
             * call to write will be appended to the sink.
             *
             * This method is not blocking beyond the transfer length bytes.
             *
             * @param in the input bytes to write out
             * @param length the length of the byte array in
             * @return length in bytes that were actually written
             */
            [[nodiscard]] virtual size_t write(const void* in, size_t length) noexcept = 0;

            /**
             * return the id of this data source
             * @return std::string representing the id of this data source
             */
            virtual std::string id() const noexcept { return ""; }

            /**
             * Read one byte.
             * @param out the byte to read to
             * @return length in bytes that was actually read and put
             * into out
             */
            size_t write_byte(uint8_t& out) noexcept;

            /**
             * Returns the output position indicator.
             *
             * @return number of bytes written so far.
             */
            virtual uint64_t tellp() const noexcept = 0;

            virtual std::string to_string() const noexcept = 0;
    };

    /**
     * File based byte output stream, including named file descriptor.
     */
    class ByteOutStream_File final : public ByteOutStream {
       private:
          jau::fs::file_stats stats;
          /**
           * We mimic std::ofstream via OS level file descriptor operations,
           * giving us more flexibility and enabling use of openat() operations.
           */
          int m_fd;

          // Remember: constexpr specifier used in a function or static data member (since C++17) declaration implies inline

       public:
          bool is_open() const noexcept override { return 0 <= m_fd; }

          size_t write(const void*, size_t) noexcept override;

          std::string id() const noexcept override { return stats.path(); }

          /**
           * Returns the file descriptor if is_open(), otherwise -1 for no file descriptor.
           *
           * @see is_open()
           */
          int fd() const noexcept { return m_fd; }

          /**
           * Construct a stream based byte output stream from filesystem path,
           * either an existing or new file.
           *
           * In case the file already exists, the underlying file offset is positioned at the end of the file.
           *
           * In case the given path is a local file URI starting with `file://`, see jau::io::uri::is_local_file_protocol(),
           * the leading `file://` is cut off and the remainder being used.
           *
           * @param path the path to the file, maybe a local file URI
           * @param mode file protection mode for a new file, otherwise ignored.
           */
          ByteOutStream_File(const std::string& path, const jau::fs::fmode_t mode = jau::fs::fmode_t::def_file_prot) noexcept;

          /**
           * Construct a stream based byte output stream from filesystem path and parent directory file descriptor,
           * either an existing or new file.
           *
           * In case the file already exists, the underlying file offset is positioned at the end of the file.
           *
           * In case the given path is a local file URI starting with `file://`, see jau::io::uri::is_local_file_protocol(),
           * the leading `file://` is cut off and the remainder being used.
           *
           * @param dirfd parent directory file descriptor
           * @param path the path to the file, maybe a local file URI
           * @param mode file protection mode for a new file, otherwise ignored.
           */
          ByteOutStream_File(const int dirfd, const std::string& path, const jau::fs::fmode_t mode = jau::fs::fmode_t::def_file_prot) noexcept;

          /**
           * Construct a stream based byte output stream by duplicating given file descriptor
           *
           * In case the given path is a local file URI starting with `file://`, see jau::io::uri::is_local_file_protocol(),
           * the leading `file://` is cut off and the remainder being used.
           *
           * @param fd file descriptor to duplicate leaving the given `fd` untouched
           */
          ByteOutStream_File(const int fd) noexcept;

          ByteOutStream_File(const ByteOutStream_File&) = delete;

          ByteOutStream_File& operator=(const ByteOutStream_File&) = delete;

          void close() noexcept override;

          ~ByteOutStream_File() noexcept override { close(); }

          uint64_t tellp() const noexcept override { return m_bytes_consumed; }

          std::string to_string() const noexcept override;

       private:
          uint64_t m_bytes_consumed;
    };


    /**@}*/

} // namespace elevator::io

#endif /* JAU_BYTE_STREAM_HPP_ */