1
Fork 0
mirror of https://github.com/RGBCube/serenity synced 2025-07-27 03:07:36 +00:00

AK: Move Stream and SeekableStream from LibCore

`Stream` will be qualified as `AK::Stream` until we remove the
`Core::Stream` namespace. `IODevice` now reuses the `SeekMode` that is
defined by `SeekableStream`, since defining its own would require us to
qualify it with `AK::SeekMode` everywhere.
This commit is contained in:
Tim Schumacher 2023-01-22 05:09:11 +01:00 committed by Andrew Kaster
parent 5f2ea31816
commit 8464da1439
96 changed files with 620 additions and 586 deletions

View file

@ -22,7 +22,7 @@ namespace Core::Stream {
/// A stream wrapper class that allows you to read arbitrary amounts of bits
/// in big-endian order from another stream.
class BigEndianInputBitStream : public Stream {
class BigEndianInputBitStream : public AK::Stream {
public:
static ErrorOr<NonnullOwnPtr<BigEndianInputBitStream>> construct(MaybeOwned<Stream> stream)
{
@ -131,7 +131,7 @@ private:
/// A stream wrapper class that allows you to read arbitrary amounts of bits
/// in little-endian order from another stream.
class LittleEndianInputBitStream : public Stream {
class LittleEndianInputBitStream : public AK::Stream {
public:
static ErrorOr<NonnullOwnPtr<LittleEndianInputBitStream>> construct(MaybeOwned<Stream> stream)
{
@ -240,7 +240,7 @@ private:
/// A stream wrapper class that allows you to write arbitrary amounts of bits
/// in big-endian order to another stream.
class BigEndianOutputBitStream : public Stream {
class BigEndianOutputBitStream : public AK::Stream {
public:
static ErrorOr<NonnullOwnPtr<BigEndianOutputBitStream>> construct(MaybeOwned<Stream> stream)
{
@ -323,7 +323,7 @@ private:
/// A stream wrapper class that allows you to write arbitrary amounts of bits
/// in little-endian order to another stream.
class LittleEndianOutputBitStream : public Stream {
class LittleEndianOutputBitStream : public AK::Stream {
public:
static ErrorOr<NonnullOwnPtr<LittleEndianOutputBitStream>> construct(MaybeOwned<Stream> stream)
{

View file

@ -176,7 +176,7 @@ ErrorOr<void> ConfigFile::sync()
return Error::from_errno(ENOENT);
TRY(m_file->truncate(0));
TRY(m_file->seek(0, Stream::SeekMode::SetPosition));
TRY(m_file->seek(0, SeekMode::SetPosition));
for (auto& it : m_groups) {
TRY(m_file->write(DeprecatedString::formatted("[{}]\n", it.key).bytes()));

View file

@ -40,7 +40,6 @@ enum class TimerShouldFireWhenNotVisible;
namespace Stream {
class File;
class Socket;
class Stream;
class BufferedSocketBase;
}

View file

@ -9,6 +9,7 @@
#include <AK/EnumBits.h>
#include <AK/Forward.h>
#include <AK/Stream.h>
#include <LibCore/Object.h>
namespace Core {
@ -62,12 +63,6 @@ enum class OpenMode : unsigned {
KeepOnExec = 32,
};
enum class SeekMode {
SetPosition,
FromCurrentPosition,
FromEndPosition,
};
AK_ENUM_BITWISE_OPERATORS(OpenMode)
class IODevice : public Object {

View file

@ -49,7 +49,7 @@ private:
/// A stream class that allows for writing to an automatically allocating memory area
/// and reading back the written data afterwards.
class AllocatingMemoryStream final : public Stream {
class AllocatingMemoryStream final : public AK::Stream {
public:
virtual ErrorOr<Bytes> read(Bytes) override;
virtual ErrorOr<size_t> write(ReadonlyBytes) override;

View file

@ -11,7 +11,7 @@
namespace Core {
NetworkJob::NetworkJob(Core::Stream::Stream& output_stream)
NetworkJob::NetworkJob(AK::Stream& output_stream)
: m_output_stream(output_stream)
{
}

View file

@ -52,7 +52,7 @@ public:
}
protected:
NetworkJob(Core::Stream::Stream&);
NetworkJob(AK::Stream&);
void did_finish(NonnullRefPtr<NetworkResponse>&&);
void did_fail(Error);
void did_progress(Optional<u32> total_size, u32 downloaded);
@ -61,7 +61,7 @@ protected:
private:
RefPtr<NetworkResponse> m_response;
Core::Stream::Stream& m_output_stream;
AK::Stream& m_output_stream;
Error m_error { Error::None };
};

View file

@ -15,9 +15,9 @@ namespace Core {
HashMap<uid_t, DeprecatedString> ProcessStatisticsReader::s_usernames;
ErrorOr<AllProcessesStatistics> ProcessStatisticsReader::get_all(Core::Stream::SeekableStream& proc_all_file, bool include_usernames)
ErrorOr<AllProcessesStatistics> ProcessStatisticsReader::get_all(SeekableStream& proc_all_file, bool include_usernames)
{
TRY(proc_all_file.seek(0, Core::Stream::SeekMode::SetPosition));
TRY(proc_all_file.seek(0, SeekMode::SetPosition));
AllProcessesStatistics all_processes_statistics;

View file

@ -72,7 +72,7 @@ struct AllProcessesStatistics {
class ProcessStatisticsReader {
public:
static ErrorOr<AllProcessesStatistics> get_all(Core::Stream::SeekableStream&, bool include_usernames = true);
static ErrorOr<AllProcessesStatistics> get_all(SeekableStream&, bool include_usernames = true);
static ErrorOr<AllProcessesStatistics> get_all(bool include_usernames = true);
private:

View file

@ -22,124 +22,6 @@
namespace Core::Stream {
ErrorOr<void> Stream::read_entire_buffer(Bytes buffer)
{
size_t nread = 0;
while (nread < buffer.size()) {
if (is_eof())
return Error::from_string_literal("Reached end-of-file before filling the entire buffer");
auto result = read(buffer.slice(nread));
if (result.is_error()) {
if (result.error().is_errno() && result.error().code() == EINTR) {
continue;
}
return result.release_error();
}
nread += result.value().size();
}
return {};
}
ErrorOr<ByteBuffer> Stream::read_until_eof(size_t block_size)
{
return read_until_eof_impl(block_size);
}
ErrorOr<ByteBuffer> Stream::read_until_eof_impl(size_t block_size, size_t expected_file_size)
{
ByteBuffer data;
data.ensure_capacity(expected_file_size);
size_t total_read = 0;
Bytes buffer;
while (!is_eof()) {
if (buffer.is_empty()) {
buffer = TRY(data.get_bytes_for_writing(block_size));
}
auto nread = TRY(read(buffer)).size();
total_read += nread;
buffer = buffer.slice(nread);
}
data.resize(total_read);
return data;
}
ErrorOr<void> Stream::discard(size_t discarded_bytes)
{
// Note: This was chosen arbitrarily.
// Note: This can't be PAGE_SIZE because it is defined to sysconf() on Lagom.
constexpr size_t continuous_read_size = 4096;
Array<u8, continuous_read_size> buffer;
while (discarded_bytes > 0) {
if (is_eof())
return Error::from_string_literal("Reached end-of-file before reading all discarded bytes");
auto slice = TRY(read(buffer.span().slice(0, min(discarded_bytes, continuous_read_size))));
discarded_bytes -= slice.size();
}
return {};
}
ErrorOr<void> Stream::write_entire_buffer(ReadonlyBytes buffer)
{
size_t nwritten = 0;
while (nwritten < buffer.size()) {
auto result = write(buffer.slice(nwritten));
if (result.is_error()) {
if (result.error().is_errno() && result.error().code() == EINTR) {
continue;
}
return result.release_error();
}
nwritten += result.value();
}
return {};
}
ErrorOr<size_t> SeekableStream::tell() const
{
// Seek with 0 and SEEK_CUR does not modify anything despite the const_cast,
// so it's safe to do this.
return const_cast<SeekableStream*>(this)->seek(0, SeekMode::FromCurrentPosition);
}
ErrorOr<size_t> SeekableStream::size()
{
auto original_position = TRY(tell());
auto seek_result = seek(0, SeekMode::FromEndPosition);
if (seek_result.is_error()) {
// Let's try to restore the original position, just in case.
auto restore_result = seek(original_position, SeekMode::SetPosition);
if (restore_result.is_error()) {
dbgln("Core::SeekableStream::size: Couldn't restore initial position, stream might have incorrect position now!");
}
return seek_result.release_error();
}
TRY(seek(original_position, SeekMode::SetPosition));
return seek_result.value();
}
ErrorOr<void> SeekableStream::discard(size_t discarded_bytes)
{
TRY(seek(discarded_bytes, SeekMode::FromCurrentPosition));
return {};
}
ErrorOr<NonnullOwnPtr<File>> File::open(StringView filename, OpenMode mode, mode_t permissions)
{
auto file = TRY(adopt_nonnull_own_or_enomem(new (nothrow) File(mode)));
@ -780,7 +662,7 @@ void WrappedAKOutputStream::close()
{
}
WrapInAKInputStream::WrapInAKInputStream(Core::Stream::Stream& stream)
WrapInAKInputStream::WrapInAKInputStream(AK::Stream& stream)
: m_stream(stream)
{
}
@ -826,7 +708,7 @@ bool WrapInAKInputStream::discard_or_error(size_t count)
return true;
}
WrapInAKOutputStream::WrapInAKOutputStream(Core::Stream::Stream& stream)
WrapInAKOutputStream::WrapInAKOutputStream(AK::Stream& stream)
: m_stream(stream)
{
}

View file

@ -17,6 +17,7 @@
#include <AK/Noncopyable.h>
#include <AK/Result.h>
#include <AK/Span.h>
#include <AK/Stream.h>
#include <AK/Time.h>
#include <AK/Variant.h>
#include <LibCore/Notifier.h>
@ -27,119 +28,6 @@
namespace Core::Stream {
/// The base, abstract class for stream operations. This class defines the
/// operations one can perform on every stream in LibCore.
/// Operations without a sensible default that are unsupported by an implementation
/// of a Stream should return EBADF as an error.
class Stream {
public:
/// Reads into a buffer, with the maximum size being the size of the buffer.
/// The amount of bytes read can be smaller than the size of the buffer.
/// Returns either the bytes that were read, or an errno in the case of
/// failure.
virtual ErrorOr<Bytes> read(Bytes) = 0;
/// Tries to fill the entire buffer through reading. Returns whether the
/// buffer was filled without an error.
virtual ErrorOr<void> read_entire_buffer(Bytes);
/// Reads the stream until EOF, storing the contents into a ByteBuffer which
/// is returned once EOF is encountered. The block size determines the size
/// of newly allocated chunks while reading.
virtual ErrorOr<ByteBuffer> read_until_eof(size_t block_size = 4096);
/// Discards the given number of bytes from the stream. As this is usually used
/// as an efficient version of `read_entire_buffer`, it returns an error
/// if reading failed or if not all bytes could be discarded.
/// Unless specifically overwritten, this just uses read() to read into an
/// internal stack-based buffer.
virtual ErrorOr<void> discard(size_t discarded_bytes);
/// Tries to write the entire contents of the buffer. It is possible for
/// less than the full buffer to be written. Returns either the amount of
/// bytes written into the stream, or an errno in the case of failure.
virtual ErrorOr<size_t> write(ReadonlyBytes) = 0;
/// Same as write, but does not return until either the entire buffer
/// contents are written or an error occurs.
virtual ErrorOr<void> write_entire_buffer(ReadonlyBytes);
template<typename T>
requires(requires(Stream& stream) { { T::read_from_stream(stream) } -> SameAs<ErrorOr<T>>; })
ErrorOr<T> read_value()
{
return T::read_from_stream(*this);
}
template<typename T>
requires(Traits<T>::is_trivially_serializable())
ErrorOr<T> read_value()
{
alignas(T) u8 buffer[sizeof(T)] = {};
TRY(read_entire_buffer({ &buffer, sizeof(buffer) }));
return bit_cast<T>(buffer);
}
template<typename T>
requires(requires(T t, Stream& stream) { { t.write_to_stream(stream) } -> SameAs<ErrorOr<void>>; })
ErrorOr<void> write_value(T const& value)
{
return value.write_to_stream(*this);
}
template<typename T>
requires(Traits<T>::is_trivially_serializable())
ErrorOr<void> write_value(T const& value)
{
return write_entire_buffer({ &value, sizeof(value) });
}
/// Returns whether the stream has reached the end of file. For sockets,
/// this most likely means that the protocol has disconnected (in the case
/// of TCP). For seekable streams, this means the end of the file. Note that
/// is_eof will only return true _after_ a read with 0 length, so this
/// method should be called after a read.
virtual bool is_eof() const = 0;
virtual bool is_open() const = 0;
virtual void close() = 0;
virtual ~Stream()
{
}
protected:
/// Provides a default implementation of read_until_eof that works for streams
/// that behave like POSIX file descriptors. expected_file_size can be
/// passed as a heuristic for what the Stream subclass expects the file
/// content size to be in order to reduce allocations (does not affect
/// actual reading).
ErrorOr<ByteBuffer> read_until_eof_impl(size_t block_size, size_t expected_file_size = 0);
};
enum class SeekMode {
SetPosition,
FromCurrentPosition,
FromEndPosition,
};
/// Adds seekability to Core::Stream. Classes inheriting from SeekableStream
/// will be seekable to any point in the stream.
class SeekableStream : public Stream {
public:
/// Seeks to the given position in the given mode. Returns either the
/// current position of the file, or an errno in the case of an error.
virtual ErrorOr<size_t> seek(i64 offset, SeekMode) = 0;
/// Returns the current position of the file, or an errno in the case of
/// an error.
virtual ErrorOr<size_t> tell() const;
/// Returns the total size of the stream, or an errno in the case of an
/// error. May not preserve the original position on the stream on failure.
virtual ErrorOr<size_t> size();
/// Shrinks or extends the stream to the given size. Returns an errno in
/// the case of an error.
virtual ErrorOr<void> truncate(off_t length) = 0;
/// Seeks until after the given amount of bytes to be discarded instead of
/// reading and discarding everything manually;
virtual ErrorOr<void> discard(size_t discarded_bytes) override;
};
enum class PreventSIGPIPE {
No,
Yes,
@ -147,7 +35,7 @@ enum class PreventSIGPIPE {
/// The Socket class is the base class for all concrete BSD-style socket
/// classes. Sockets are non-seekable streams which can be read byte-wise.
class Socket : public Stream {
class Socket : public AK::Stream {
public:
Socket(Socket&&) = default;
Socket& operator=(Socket&&) = default;
@ -577,7 +465,7 @@ private:
// Buffered stream wrappers
template<typename T>
concept StreamLike = IsBaseOf<Stream, T>;
concept StreamLike = IsBaseOf<AK::Stream, T>;
template<typename T>
concept SeekableStreamLike = IsBaseOf<SeekableStream, T>;
template<typename T>
@ -1028,7 +916,7 @@ using ReusableTCPSocket = BasicReusableSocket<TCPSocket>;
using ReusableUDPSocket = BasicReusableSocket<UDPSocket>;
// Note: This is only a temporary hack, to break up the task of moving away from AK::Stream into smaller parts.
class WrappedAKInputStream final : public Stream {
class WrappedAKInputStream final : public AK::Stream {
public:
WrappedAKInputStream(NonnullOwnPtr<DeprecatedInputStream> stream);
virtual ErrorOr<Bytes> read(Bytes) override;
@ -1043,7 +931,7 @@ private:
};
// Note: This is only a temporary hack, to break up the task of moving away from AK::Stream into smaller parts.
class WrappedAKOutputStream final : public Stream {
class WrappedAKOutputStream final : public AK::Stream {
public:
WrappedAKOutputStream(NonnullOwnPtr<DeprecatedOutputStream> stream);
virtual ErrorOr<Bytes> read(Bytes) override;
@ -1059,25 +947,25 @@ private:
// Note: This is only a temporary hack, to break up the task of moving away from AK::Stream into smaller parts.
class WrapInAKInputStream final : public DeprecatedInputStream {
public:
WrapInAKInputStream(Core::Stream::Stream& stream);
WrapInAKInputStream(AK::Stream& stream);
virtual size_t read(Bytes) override;
virtual bool unreliable_eof() const override;
virtual bool read_or_error(Bytes) override;
virtual bool discard_or_error(size_t count) override;
private:
Core::Stream::Stream& m_stream;
AK::Stream& m_stream;
};
// Note: This is only a temporary hack, to break up the task of moving away from AK::Stream into smaller parts.
class WrapInAKOutputStream final : public DeprecatedOutputStream {
public:
WrapInAKOutputStream(Core::Stream::Stream& stream);
WrapInAKOutputStream(AK::Stream& stream);
virtual size_t write(ReadonlyBytes) override;
virtual bool write_or_error(ReadonlyBytes) override;
private:
Core::Stream::Stream& m_stream;
AK::Stream& m_stream;
};
}