ObjFW  Check-in [481225349f]

	if (ret == 0)
		_atEndOfStream = true;

	return ret;
}

- (size_t)lowlevelWriteBuffer: (const void *)buffer
		       length: (size_t)length
{
	if (_handle == OF_INVALID_FILE_HANDLE)
		@throw [OFNotOpenException exceptionWithObject: self];

#if defined(OF_WINDOWS)
	int bytesWritten;

	if ((bytesWritten = write(_handle, buffer, length)) < 0)
		@throw [OFWriteFailedException exceptionWithObject: self
						   requestedLength: length
						      bytesWritten: 0
							     errNo: errno];
#endif

	return (size_t)bytesWritten;




}

- (of_offset_t)lowlevelSeekToOffset: (of_offset_t)offset
			     whence: (int)whence
{
	of_offset_t ret;

	_headersSent = true;
	_chunked = [[headers objectForKey: @"Transfer-Encoding"]
	    isEqual: @"chunked"];

	objc_autoreleasePoolPop(pool);
}

- (size_t)lowlevelWriteBuffer: (const void *)buffer
		       length: (size_t)length
{
	/* TODO: Use non-blocking writes */

	void *pool;

	if (_socket == nil)
		@throw [OFNotOpenException exceptionWithObject: self];

	if (!_headersSent)
		[self of_sendHeaders];

	if (!_chunked)
		return [_socket writeBuffer: buffer
				     length: length];



	pool = objc_autoreleasePoolPush();
	[_socket writeString: [OFString stringWithFormat: @"%zx\r\n", length]];
	objc_autoreleasePoolPop(pool);

	[_socket writeBuffer: buffer
		      length: length];
	[_socket writeBuffer: "\r\n"
		      length: 2];

	return length;
}

- (void)close
{
	if (_socket == nil)
		@throw [OFNotOpenException exceptionWithObject: self];

	if (ret == 0)
		_atEndOfStream = true;

	return ret;
}

- (size_t)lowlevelWriteBuffer: (const void *)buffer
		       length: (size_t)length
{
#ifndef OF_WINDOWS
	ssize_t bytesWritten;

	if (_writePipe[1] == -1)
		@throw [OFNotOpenException exceptionWithObject: self];

		@throw [OFWriteFailedException exceptionWithObject: self
						   requestedLength: length
						      bytesWritten: 0
							     errNo: errNo];
	}
#endif

	return (size_t)bytesWritten;




}

- (int)fileDescriptorForReading
{
#ifndef OF_WINDOWS
	return _readPipe[0];
#else

			selector: (SEL)selector
			 context: (nullable id)context;
+ (void)of_addAsyncReadLineForStream: (OFStream *)stream
			    encoding: (of_string_encoding_t)encoding
			      target: (id)target
			    selector: (SEL)selector
			     context: (nullable id)context;
+ (void)of_addAsyncWriteForStream: (OFStream *)stream
			   buffer: (const void *)buffer
			   length: (size_t)length
			   target: (id)target
			 selector: (SEL)selector
			  context: (nullable id)context;
+ (void)of_addAsyncAcceptForTCPSocket: (OFTCPSocket *)socket
			       target: (id)target
			     selector: (SEL)selector
			      context: (nullable id)context;
+ (void)of_addAsyncReceiveForUDPSocket: (OFUDPSocket *)socket
				buffer: (void *)buffer
				length: (size_t)length

+ (void)of_addAsyncReadForStream: (OFStream *)stream
			  buffer: (void *)buffer
		     exactLength: (size_t)length
			   block: (of_stream_async_read_block_t)block;
+ (void)of_addAsyncReadLineForStream: (OFStream *)stream
			    encoding: (of_string_encoding_t)encoding
			       block: (of_stream_async_read_line_block_t)block;
+ (void)of_addAsyncWriteForStream: (OFStream *)stream
			   buffer: (const void *)buffer
			   length: (size_t)length
			    block: (of_stream_async_write_block_t)block;
+ (void)of_addAsyncAcceptForTCPSocket: (OFTCPSocket *)socket
				block: (of_tcp_socket_async_accept_block_t)
					   block;
+ (void)of_addAsyncReceiveForUDPSocket: (OFUDPSocket *)socket
				buffer: (void *)buffer
				length: (size_t)length
				 block: (of_udp_socket_async_receive_block_t)

{
	OFSortedList *_timersQueue;
#ifdef OF_HAVE_THREADS
	OFMutex *_timersQueueLock;
#endif
#if defined(OF_HAVE_SOCKETS)
	OFKernelEventObserver *_kernelEventObserver;
	OFMutableDictionary *_readQueues, *_writeQueues;
#elif defined(OF_HAVE_THREADS)
	OFCondition *_condition;
#endif
	volatile bool _stop;
}

/*!

{
@public
	id _target;
	SEL _selector;
	id _context;
}

- (bool)handleObject: (id)object;
@end

@interface OFRunLoop_ReadQueueItem: OFRunLoop_QueueItem
{
@public
# ifdef OF_HAVE_BLOCKS
	of_stream_async_read_block_t _block;

@public
# ifdef OF_HAVE_BLOCKS
	of_stream_async_read_line_block_t _block;
# endif
	of_string_encoding_t _encoding;
}
@end

@interface OFRunLoop_WriteQueueItem: OFRunLoop_QueueItem
{
@public
# ifdef OF_HAVE_BLOCKS
	of_stream_async_write_block_t _block;
# endif
	const void *_buffer;
	size_t _length, _writtenLength;
}
@end

@interface OFRunLoop_AcceptQueueItem: OFRunLoop_QueueItem
{
@public
# ifdef OF_HAVE_BLOCKS
	of_tcp_socket_async_accept_block_t _block;
# endif

# endif
	void *_buffer;
	size_t _length;
}
@end

@implementation OFRunLoop_QueueItem
- (bool)handleObject: (id)object
{
	OF_UNRECOGNIZED_SELECTOR
}

- (void)dealloc
{
	[_target release];
	[_context release];

	[super dealloc];
}
@end

@implementation OFRunLoop_ReadQueueItem
- (bool)handleObject: (id)object
{
	size_t length;
	id exception = nil;

	@try {
		length = [object readIntoBuffer: _buffer
					 length: _length];

	[super dealloc];
}
# endif
@end

@implementation OFRunLoop_ExactReadQueueItem
- (bool)handleObject: (id)object
{
	size_t length;
	id exception = nil;

	@try {
		length = [object readIntoBuffer: (char *)_buffer + _readLength
					 length: _exactLength - _readLength];

	[super dealloc];
}
# endif
@end

@implementation OFRunLoop_ReadLineQueueItem
- (bool)handleObject: (id)object
{
	OFString *line;
	id exception = nil;

	@try {
		line = [object tryReadLineWithEncoding: _encoding];
	} @catch (id e) {

{
	[_block release];

	[super dealloc];
}
# endif
@end

@implementation OFRunLoop_WriteQueueItem
# ifdef OF_HAVE_BLOCKS
- (bool)handleObject: (id)object
{
	size_t length;
	id exception = nil;

	@try {
		length = [object writeBuffer: (char *)_buffer + _writtenLength
				      length: _length - _writtenLength];
	} @catch (id e) {
		length = 0;
		exception = e;
	}

	_writtenLength += length;

	if (_writtenLength != _length && exception == nil)
		return true;

# ifdef OF_HAVE_BLOCKS
	if (_block != NULL)
		_block(object, _buffer, _writtenLength, exception);
	else {
# endif
		void (*func)(id, SEL, OFStream *, const void *, size_t, id,
		    id) = (void (*)(id, SEL, OFStream *, const void *, size_t,
		    id, id))[_target methodForSelector: _selector];

		func(_target, _selector, object, _buffer, _writtenLength,
		    _context, exception);
# ifdef OF_HAVE_BLOCKS
	}
# endif

	return false;
}

- (void)dealloc
{
	[_block release];

	[super dealloc];
}
# endif
@end

@implementation OFRunLoop_AcceptQueueItem
- (bool)handleObject: (id)object
{
	OFTCPSocket *newSocket;
	id exception = nil;

	@try {
		newSocket = [object accept];
	} @catch (id e) {

	[super dealloc];
}
# endif
@end

@implementation OFRunLoop_UDPReceiveQueueItem
- (bool)handleObject: (id)object
{
	size_t length;
	of_udp_socket_address_t address;
	id exception = nil;

	@try {
		length = [object receiveIntoBuffer: _buffer

		    addObjectForReading: object];			\
									\
	queueItem = [[[type alloc] init] autorelease];			\
	code								\
	[queue appendObject: queueItem];				\
									\
	objc_autoreleasePoolPop(pool);
# define ADD_WRITE(type, object, code)					\
	void *pool = objc_autoreleasePoolPush();			\
	OFRunLoop *runLoop = [self currentRunLoop];			\
	OFList *queue = [runLoop->_writeQueues objectForKey: object];	\
	type *queueItem;						\
									\
	if (queue == nil) {						\
		queue = [OFList list];					\
		[runLoop->_writeQueues setObject: queue			\
					  forKey: object];		\
	}								\
									\
	if ([queue count] == 0)						\
		[runLoop->_kernelEventObserver				\
		    addObjectForWriting: object];			\
									\
	queueItem = [[[type alloc] init] autorelease];			\
	code								\
	[queue appendObject: queueItem];				\
									\
	objc_autoreleasePoolPop(pool);

+ (void)of_addAsyncReadForStream: (OFStream *)stream
			  buffer: (void *)buffer
			  length: (size_t)length
			  target: (id)target
			selector: (SEL)selector
			 context: (id)context

	ADD_READ(OFRunLoop_ReadLineQueueItem, stream, {
		queueItem->_target = [target retain];
		queueItem->_selector = selector;
		queueItem->_context = [context retain];
		queueItem->_encoding = encoding;
	})
}

+ (void)of_addAsyncWriteForStream: (OFStream *)stream
			   buffer: (const void *)buffer
			   length: (size_t)length
			   target: (id)target
			 selector: (SEL)selector
			  context: (id)context
{
	ADD_WRITE(OFRunLoop_WriteQueueItem, stream, {
		queueItem->_target = [target retain];
		queueItem->_selector = selector;
		queueItem->_context = [context retain];
		queueItem->_buffer = buffer;
		queueItem->_length = length;
	})
}

+ (void)of_addAsyncAcceptForTCPSocket: (OFTCPSocket *)stream
			       target: (id)target
			     selector: (SEL)selector
			      context: (id)context
{
	ADD_READ(OFRunLoop_AcceptQueueItem, stream, {

			       block: (of_stream_async_read_line_block_t)block
{
	ADD_READ(OFRunLoop_ReadLineQueueItem, stream, {
		queueItem->_block = [block copy];
		queueItem->_encoding = encoding;
	})
}

+ (void)of_addAsyncWriteForStream: (OFStream *)stream
			   buffer: (const void *)buffer
			   length: (size_t)length
			    block: (of_stream_async_write_block_t)block
{
	ADD_WRITE(OFRunLoop_WriteQueueItem, stream, {
		queueItem->_block = [block copy];
		queueItem->_buffer = buffer;
		queueItem->_length = length;
	})
}

+ (void)of_addAsyncAcceptForTCPSocket: (OFTCPSocket *)stream
				block: (of_tcp_socket_async_accept_block_t)block
{
	ADD_READ(OFRunLoop_AcceptQueueItem, stream, {
		queueItem->_block = [block copy];
	})

		queueItem->_buffer = buffer;
		queueItem->_length = length;
		queueItem->_block = [block copy];
	})
}
# endif
# undef ADD_READ
# undef ADD_WRITE

+ (void)of_cancelAsyncRequestsForObject: (id)object
{
	void *pool = objc_autoreleasePoolPush();
	OFRunLoop *runLoop = [self currentRunLoop];
	OFList *queue;

	if ((queue = [runLoop->_writeQueues objectForKey: object]) != nil) {
		assert([queue count] > 0);

		/*
		 * Clear the queue now, in case this has been called from a
		 * handler, as otherwise, we'd do the cleanups below twice.
		 */
		[queue removeAllObjects];

		[runLoop->_kernelEventObserver removeObjectForWriting: object];
		[runLoop->_writeQueues removeObjectForKey: object];
	}

	if ((queue = [runLoop->_readQueues objectForKey: object]) != nil) {
		assert([queue count] > 0);

		/*
		 * Clear the queue now, in case this has been called from a
		 * handler, as otherwise, we'd do the cleanups below twice.

#endif

#if defined(OF_HAVE_SOCKETS)
		_kernelEventObserver = [[OFKernelEventObserver alloc] init];
		[_kernelEventObserver setDelegate: self];

		_readQueues = [[OFMutableDictionary alloc] init];
		_writeQueues = [[OFMutableDictionary alloc] init];
#elif defined(OF_HAVE_THREADS)
		_condition = [[OFCondition alloc] init];
#endif
	} @catch (id e) {
		[self release];
		@throw e;
	}

	return self;
}

- (void)dealloc
{
	[_timersQueue release];
#ifdef OF_HAVE_THREADS
	[_timersQueueLock release];
#endif
#if defined(OF_HAVE_SOCKETS)
	[_kernelEventObserver release];
	[_readQueues release];
	[_writeQueues release];
#elif defined(OF_HAVE_THREADS)
	[_condition release];
#endif

	[super dealloc];
}

	 */
	OFList OF_GENERIC(OF_KINDOF(OFRunLoop_ReadQueueItem *)) *queue =
	    [[_readQueues objectForKey: object] retain];

	assert(queue != nil);

	@try {
		if (![[queue firstObject] handleObject: object]) {
			of_list_object_t *listObject = [queue firstListObject];

			/*
			 * The handler might have called -[cancelAsyncRequests]
			 * so that our queue is now empty, in which case we
			 * should do nothing.
			 */
			if (listObject != NULL) {
				[queue removeListObject: listObject];

				if ([queue count] == 0) {
					[_kernelEventObserver
					    removeObjectForReading: object];
					[_readQueues
					    removeObjectForKey: object];
				}
			}
		}
	} @finally {
		[queue release];
	}
}

- (void)objectIsReadyForWriting: (id)object
{
	/*
	 * Retain the queue so that it doesn't disappear from us because the
	 * handler called -[cancelAsyncRequests].
	 */
	OFList OF_GENERIC(OF_KINDOF(OFRunLoop_WriteQueueItem *)) *queue =
	    [[_writeQueues objectForKey: object] retain];

	assert(queue != nil);

	@try {
		if (![[queue firstObject] handleObject: object]) {
			of_list_object_t *listObject = [queue firstListObject];

			/*
			 * The handler might have called -[cancelAsyncRequests]
			 * so that our queue is now empty, in which case we
			 * should do nothing.
			 */
			if (listObject != NULL) {
				[queue removeListObject: listObject];

				if ([queue count] == 0) {
					[_kernelEventObserver
					    removeObjectForWriting: object];
					[_writeQueues
					    removeObjectForKey: object];
				}
			}
		}
	} @finally {
		[queue release];
	}
}
#endif

	if (ret == 0)
		_atEndOfStream = true;

	return ret;
}

- (size_t)lowlevelWriteBuffer: (const void *)buffer
		       length: (size_t)length
{
#ifndef OF_MORPHOS
	if (_fd == -1)
		@throw [OFNotOpenException exceptionWithObject: self];

# ifndef OF_WINDOWS
	ssize_t bytesWritten;

	if ((bytesWritten = Write(_handle, (void *)buffer, length)) < 0)
		@throw [OFWriteFailedException exceptionWithObject: self
						   requestedLength: length
						      bytesWritten: 0
							     errNo: EIO];
#endif

	return (size_t)bytesWritten;




}

#if !defined(OF_WINDOWS) && !defined(OF_MORPHOS)
- (int)fileDescriptorForReading
{
	return _fd;
}

	}

	objc_autoreleasePoolPop(pool);

	return j;
}

- (size_t)lowlevelWriteBuffer: (const void *)buffer_
		       length: (size_t)length
{
	const char *buffer = buffer_;
	char16_t *tmp;
	size_t i = 0, j = 0;

	if (length > SIZE_MAX / 2)
		@throw [OFOutOfRangeException exception];

			    exceptionWithObject: self
				requestedLength: j * 2
				   bytesWritten: bytesWritten * 2
					  errNo: 0];
	} @finally {
		[self freeMemory: tmp];
	}

	/*
	 * We do not count in bytes when writing to the Win32 console. But
	 * since any incomplete write is an exception here anyway, we can just
	 * return length.
	 */
	return length;
}
@end

 *	       occurred
 * @param exception An exception which occurred while reading or `nil` on
 *		    success
 * @return A bool whether the same block should be used for the next read
 */
typedef bool (^of_stream_async_read_line_block_t)(OFStream *stream,
    OFString *_Nullable line, id _Nullable exception);

/*!
 * @brief A block which is called when data was written to the stream.
 *
 * @param stream The stream to which data was written
 * @param buffer The buffer which was written to the stream
 * @param length The length of the data that bas been written
 * @param exception An exception which occurred while reading or `nil` on
 *		    success
 */
typedef void (^of_stream_async_write_block_t)(OFStream *stream,
    const void *buffer, size_t length, id _Nullable exception);
#endif

/*!
 * @class OFStream OFStream.h ObjFW/OFStream.h
 *
 * @brief A base class for different types of streams.
 *

- (void)flushWriteBuffer;

/*!
 * @brief Writes from a buffer into the stream.
 *
 * @param buffer The buffer from which the data is written into the stream
 * @param length The length of the data that should be written
 * @return The number of bytes written. This can only differ from the specified
 *	   length in non-blocking mode.
 */
- (size_t)writeBuffer: (const void *)buffer
	       length: (size_t)length;

#ifdef OF_HAVE_SOCKETS
/*!
 * @brief Asynchronously writes a buffer into the stream.
 *
 * @note The stream must implement @ref fileDescriptorForWriting and return a
 *	 valid file descriptor in order for this to work!
 *
 * @param buffer The buffer from which the data is written into the stream. The
 *		 buffer needs to be valid until the write request is completed!
 * @param length The length of the data that should be written
 * @param target The target on which the selector should be called when the data
 *		 has been written.
 * @param selector The selector to call on the target. The signature must be
 *		   `void (OFStream *stream, const void *buffer, size_t length,
 *		   id context, id exception)`.
 */
- (void)asyncWriteBuffer: (const void *)buffer
		  length: (size_t)length
		  target: (id)target
		selector: (SEL)selector
		 context: (nullable id)context;

# ifdef OF_HAVE_BLOCKS
/*!
 * @brief Asynchronously writes a buffer into the stream.
 *
 * @note The stream must implement @ref fileDescriptorForWriting and return a
 *	 valid file descriptor in order for this to work!
 *
 * @param buffer The buffer from which the data is written into the stream. The
 *		 buffer needs to be valid until the write request is completed!
 * @param length The length of the data that should be written
 * @param block The block to call when the data has been written
 */
- (void)asyncWriteBuffer: (const void *)buffer
		  length: (size_t)length
		   block: (of_stream_async_write_block_t)block;
# endif
#endif

/*!
 * @brief Writes a uint8_t into the stream.
 *
 * @param int8 A uint8_t
 */
- (void)writeInt8: (uint8_t)int8;

 * @warning Do not call this directly!
 *
 * @note Override this method with your actual write implementation when
 *	 subclassing!
 *
 * @param buffer The buffer with the data to write
 * @param length The length of the data to write
 * @return The number of bytes written
 */
- (size_t)lowlevelWriteBuffer: (const void *)buffer
		       length: (size_t)length;

/*!
 * @brief Returns whether the lowlevel is at the end of the stream.
 *
 * @warning Do not call this directly!
 *
 * @note Override this method with your actual end of stream checking

#import "OFInvalidArgumentException.h"
#import "OFInvalidFormatException.h"
#import "OFNotImplementedException.h"
#import "OFOutOfMemoryException.h"
#import "OFOutOfRangeException.h"
#import "OFSetOptionFailedException.h"
#import "OFTruncatedDataException.h"
#import "OFWriteFailedException.h"

#import "of_asprintf.h"

#define MIN_READ_SIZE 512

@implementation OFStream
@synthesize of_waitingForDelimiter = _waitingForDelimiter;

- (size_t)lowlevelReadIntoBuffer: (void *)buffer
			  length: (size_t)length
{
	OF_UNRECOGNIZED_SELECTOR
}

- (size_t)lowlevelWriteBuffer: (const void *)buffer
		       length: (size_t)length
{
	OF_UNRECOGNIZED_SELECTOR
}

- copy
{
	return [self retain];

			   length: _writeBufferLength];

	[self freeMemory: _writeBuffer];
	_writeBuffer = NULL;
	_writeBufferLength = 0;
}

- (size_t)writeBuffer: (const void *)buffer
	       length: (size_t)length
{
	if (!_writeBuffered) {
		size_t bytesWritten = [self lowlevelWriteBuffer: buffer
							 length: length];

		if (_blocking && bytesWritten < length)
			@throw [OFWriteFailedException
			    exceptionWithObject: self
				requestedLength: length
				   bytesWritten: bytesWritten
					  errNo: 0];

		return bytesWritten;
	} else {
		_writeBuffer = [self resizeMemory: _writeBuffer
					     size: _writeBufferLength + length];
		memcpy(_writeBuffer + _writeBufferLength, buffer, length);
		_writeBufferLength += length;

		return length;
	}
}

#ifdef OF_HAVE_SOCKETS
- (void)asyncWriteBuffer: (const void *)buffer
		  length: (size_t)length
		  target: (id)target
		selector: (SEL)selector
		 context: (id)context
{
	[OFRunLoop of_addAsyncWriteForStream: self
				      buffer: buffer
				      length: length
				      target: target
				    selector: selector
				     context: context];
}

# ifdef OF_HAVE_BLOCKS
- (void)asyncWriteBuffer: (const void *)buffer
		  length: (size_t)length
		   block: (of_stream_async_write_block_t)block
{
	[OFRunLoop of_addAsyncWriteForStream: self
				      buffer: buffer
				      length: length
				       block: block];
}
# endif
#endif

- (void)writeInt8: (uint8_t)int8
{
	[self writeBuffer: (char *)&int8
		   length: 1];
}

	if (ret == 0)
		_atEndOfStream = true;

	return ret;
}

- (size_t)lowlevelWriteBuffer: (const void *)buffer
		       length: (size_t)length
{
	if (_socket == INVALID_SOCKET)
		@throw [OFNotOpenException exceptionWithObject: self];

#ifndef OF_WINDOWS
	ssize_t bytesWritten;

		@throw [OFWriteFailedException
		    exceptionWithObject: self
			requestedLength: length
			   bytesWritten: 0
				  errNo: of_socket_errno()];
#endif

	return (size_t)bytesWritten;




}

#ifdef OF_WINDOWS
- (void)setBlocking: (bool)enable
{
	u_long v = enable;
	_blocking = enable;

	[self close];

	[_entry release];

	[super dealloc];
}

- (size_t)lowlevelWriteBuffer: (const void *)buffer
		       length: (size_t)length
{
	size_t bytesWritten;

	if (_stream == nil)
		@throw [OFNotOpenException exceptionWithObject: self];

	if ((uint64_t)length > _toWrite)
		@throw [OFOutOfRangeException exception];

	@try {
		bytesWritten = [_stream writeBuffer: buffer
					     length: length];
	} @catch (OFWriteFailedException *e) {
		_toWrite -= [e bytesWritten];
		@throw e;
	}

	_toWrite -= bytesWritten;

	return bytesWritten;
}

- (bool)lowlevelIsAtEndOfStream
{
	if (_stream == nil)
		@throw [OFNotOpenException exceptionWithObject: self];

	[_stream release];
	[_entry release];

	[super dealloc];
}

- (size_t)lowlevelWriteBuffer: (const void *)buffer
		       length: (size_t)length
{
	size_t bytesWritten;

	if ((sizeof(length) >= sizeof(int64_t) && length > INT64_MAX) ||
	    INT64_MAX - _bytesWritten < (int64_t)length)
		@throw [OFOutOfRangeException exception];

	bytesWritten = [_stream writeBuffer: buffer
				     length: length];

	_bytesWritten += (int64_t)bytesWritten;
	_CRC32 = of_crc32(_CRC32, buffer, length);

	return bytesWritten;
}

- (void)close
{
	if (_stream == nil)
		return;