/*
* Copyright (c) 2008-2021 Jonathan Schleifer <js@nil.im>
*
* All rights reserved.
*
* This file is part of ObjFW. It may be distributed under the terms of the
* Q Public License 1.0, which can be found in the file LICENSE.QPL included in
* the packaging of this file.
*
* Alternatively, it may be distributed under the terms of the GNU General
* Public License, either version 2 or 3, which can be found in the file
* LICENSE.GPLv2 or LICENSE.GPLv3 respectively included in the packaging of this
* file.
*/
#include "config.h"
#include <assert.h>
#include <errno.h>
#import "OFRunLoop.h"
#import "OFRunLoop+Private.h"
#import "OFArray.h"
#import "OFData.h"
#import "OFDictionary.h"
#ifdef OF_HAVE_SOCKETS
# import "OFKernelEventObserver.h"
# import "OFDatagramSocket.h"
# import "OFSequencedPacketSocket.h"
# import "OFSequencedPacketSocket+Private.h"
# import "OFStreamSocket.h"
# import "OFStreamSocket+Private.h"
#endif
#import "OFThread.h"
#ifdef OF_HAVE_THREADS
# import "OFMutex.h"
# import "OFCondition.h"
#endif
#import "OFSortedList.h"
#import "OFTimer.h"
#import "OFTimer+Private.h"
#import "OFDate.h"
#import "OFObserveFailedException.h"
const of_run_loop_mode_t of_run_loop_mode_default = @"of_run_loop_mode_default";
static OFRunLoop *mainRunLoop = nil;
@interface OFRunLoopState: OFObject
#ifdef OF_HAVE_SOCKETS
<OFKernelEventObserverDelegate>
#endif
{
@public
OFSortedList OF_GENERIC(OFTimer *) *_timersQueue;
#ifdef OF_HAVE_THREADS
OFMutex *_timersQueueMutex;
#endif
#if defined(OF_HAVE_SOCKETS)
OFKernelEventObserver *_kernelEventObserver;
OFMutableDictionary *_readQueues, *_writeQueues;
#elif defined(OF_HAVE_THREADS)
OFCondition *_condition;
# ifdef OF_AMIGAOS
ULONG _execSignalMask;
# endif
#endif
#ifdef OF_AMIGAOS
OFMutableData *_execSignals;
OFMutableArray *_execSignalsTargets;
OFMutableData *_execSignalsSelectors;
# ifdef OF_HAVE_THREADS
OFMutex *_execSignalsMutex;
# endif
#endif
}
@end
#ifdef OF_HAVE_SOCKETS
@interface OFRunLoopQueueItem: OFObject
{
@public
id _delegate;
}
- (bool)handleObject: (id)object;
@end
@interface OFRunLoopReadQueueItem: OFRunLoopQueueItem
{
@public
# ifdef OF_HAVE_BLOCKS
of_stream_async_read_block_t _block;
# endif
void *_buffer;
size_t _length;
}
@end
@interface OFRunLoopExactReadQueueItem: OFRunLoopQueueItem
{
@public
# ifdef OF_HAVE_BLOCKS
of_stream_async_read_block_t _block;
# endif
void *_buffer;
size_t _exactLength, _readLength;
}
@end
@interface OFRunLoopReadLineQueueItem: OFRunLoopQueueItem
{
@public
# ifdef OF_HAVE_BLOCKS
of_stream_async_read_line_block_t _block;
# endif
of_string_encoding_t _encoding;
}
@end
@interface OFRunLoopWriteDataQueueItem: OFRunLoopQueueItem
{
@public
# ifdef OF_HAVE_BLOCKS
of_stream_async_write_data_block_t _block;
# endif
OFData *_data;
size_t _writtenLength;
}
@end
@interface OFRunLoopWriteStringQueueItem: OFRunLoopQueueItem
{
@public
# ifdef OF_HAVE_BLOCKS
of_stream_async_write_string_block_t _block;
# endif
OFString *_string;
of_string_encoding_t _encoding;
size_t _writtenLength;
}
@end
# if !defined(OF_WII) && !defined(OF_NINTENDO_3DS)
@interface OFRunLoopConnectQueueItem: OFRunLoopQueueItem
@end
# endif
@interface OFRunLoopAcceptQueueItem: OFRunLoopQueueItem
{
@public
# ifdef OF_HAVE_BLOCKS
id _block;
# endif
}
@end
@interface OFRunLoopDatagramReceiveQueueItem: OFRunLoopQueueItem
{
@public
# ifdef OF_HAVE_BLOCKS
of_datagram_socket_async_receive_block_t _block;
# endif
void *_buffer;
size_t _length;
}
@end
@interface OFRunLoopDatagramSendQueueItem: OFRunLoopQueueItem
{
@public
# ifdef OF_HAVE_BLOCKS
of_datagram_socket_async_send_data_block_t _block;
# endif
OFData *_data;
of_socket_address_t _receiver;
}
@end
@interface OFRunLoopPacketReceiveQueueItem: OFRunLoopQueueItem
{
@public
# ifdef OF_HAVE_BLOCKS
of_sequenced_packet_socket_async_receive_block_t _block;
# endif
void *_buffer;
size_t _length;
}
@end
@interface OFRunLoopPacketSendQueueItem: OFRunLoopQueueItem
{
@public
# ifdef OF_HAVE_BLOCKS
of_sequenced_packet_socket_async_send_data_block_t _block;
# endif
OFData *_data;
}
@end
#endif
@implementation OFRunLoopState
- (instancetype)init
{
self = [super init];
@try {
_timersQueue = [[OFSortedList alloc] init];
#ifdef OF_HAVE_THREADS
_timersQueueMutex = [[OFMutex alloc] init];
#endif
#if defined(OF_HAVE_SOCKETS)
_kernelEventObserver = [[OFKernelEventObserver alloc] init];
_kernelEventObserver.delegate = self;
_readQueues = [[OFMutableDictionary alloc] init];
_writeQueues = [[OFMutableDictionary alloc] init];
#elif defined(OF_HAVE_THREADS)
_condition = [[OFCondition alloc] init];
#endif
#ifdef OF_AMIGAOS
_execSignals = [[OFMutableData alloc]
initWithItemSize: sizeof(ULONG)];
_execSignalsTargets = [[OFMutableArray alloc] init];
_execSignalsSelectors = [[OFMutableData alloc]
initWithItemSize: sizeof(SEL)];
# ifdef OF_HAVE_THREADS
_execSignalsMutex = [[OFMutex alloc] init];
# endif
#endif
} @catch (id e) {
[self release];
@throw e;
}
return self;
}
- (void)dealloc
{
[_timersQueue release];
#ifdef OF_HAVE_THREADS
[_timersQueueMutex release];
#endif
#if defined(OF_HAVE_SOCKETS)
[_kernelEventObserver release];
[_readQueues release];
[_writeQueues release];
#elif defined(OF_HAVE_THREADS)
[_condition release];
#endif
#ifdef OF_AMIGAOS
[_execSignals release];
[_execSignalsTargets release];
[_execSignalsSelectors release];
# ifdef OF_HAVE_THREADS
[_execSignalsMutex release];
# endif
#endif
[super dealloc];
}
#ifdef OF_HAVE_SOCKETS
- (void)objectIsReadyForReading: (id)object
{
/*
* Retain the queue so that it doesn't disappear from us because the
* handler called -[cancelAsyncRequests].
*/
OFList OF_GENERIC(OF_KINDOF(OFRunLoopReadQueueItem *)) *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) {
/*
* Make sure we keep the target until after we
* are done removing the object. The reason for
* this is that the target might call
* -[cancelAsyncRequests] in its dealloc.
*/
[[listObject->object retain] autorelease];
[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 *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) {
/*
* Make sure we keep the target until after we
* are done removing the object. The reason for
* this is that the target might call
* -[cancelAsyncRequests] in its dealloc.
*/
[[listObject->object retain] autorelease];
[queue removeListObject: listObject];
if (queue.count == 0) {
[_kernelEventObserver
removeObjectForWriting: object];
[_writeQueues
removeObjectForKey: object];
}
}
}
} @finally {
[queue release];
}
}
#endif
#ifdef OF_AMIGAOS
- (void)execSignalWasReceived: (ULONG)signalMask
{
void *pool = objc_autoreleasePoolPush();
OFData *signals;
OFArray *targets;
OFData *selectors;
const ULONG *signalsItems;
const id *targetsObjects;
const SEL *selectorsItems;
size_t count;
# ifdef OF_HAVE_THREADS
[_execSignalsMutex lock];
@try {
# endif
/*
* Create copies, so that signal handlers are allowed to modify
* signals.
*/
signals = [[_execSignals copy] autorelease];
targets = [[_execSignalsTargets copy] autorelease];
selectors = [[_execSignalsSelectors copy] autorelease];
# ifdef OF_HAVE_THREADS
} @finally {
[_execSignalsMutex unlock];
}
# endif
signalsItems = signals.items;
targetsObjects = targets.objects;
selectorsItems = selectors.items;
count = signals.count;
for (size_t i = 0; i < count; i++) {
if (signalMask & (1ul << signalsItems[i])) {
void (*callback)(id, SEL, ULONG) =
(void (*)(id, SEL, ULONG))[targetsObjects[i]
methodForSelector: selectorsItems[i]];
callback(targetsObjects[i], selectorsItems[i],
signalsItems[i]);
}
}
objc_autoreleasePoolPop(pool);
}
#endif
@end
#ifdef OF_HAVE_SOCKETS
@implementation OFRunLoopQueueItem
- (bool)handleObject: (id)object
{
OF_UNRECOGNIZED_SELECTOR
}
- (void)dealloc
{
[_delegate release];
[super dealloc];
}
@end
@implementation OFRunLoopReadQueueItem
- (bool)handleObject: (id)object
{
size_t length;
id exception = nil;
@try {
length = [object readIntoBuffer: _buffer length: _length];
} @catch (id e) {
length = 0;
exception = e;
}
# ifdef OF_HAVE_BLOCKS
if (_block != NULL)
return _block(length, exception);
else {
# endif
if (![_delegate respondsToSelector:
@selector(stream:didReadIntoBuffer:length:exception:)])
return false;
return [_delegate stream: object
didReadIntoBuffer: _buffer
length: length
exception: exception];
# ifdef OF_HAVE_BLOCKS
}
# endif
}
# ifdef OF_HAVE_BLOCKS
- (void)dealloc
{
[_block release];
[super dealloc];
}
# endif
@end
@implementation OFRunLoopExactReadQueueItem
- (bool)handleObject: (id)object
{
size_t length;
id exception = nil;
@try {
length = [object readIntoBuffer: (char *)_buffer + _readLength
length: _exactLength - _readLength];
} @catch (id e) {
length = 0;
exception = e;
}
_readLength += length;
if (_readLength != _exactLength && ![object isAtEndOfStream] &&
exception == nil)
return true;
# ifdef OF_HAVE_BLOCKS
if (_block != NULL) {
if (!_block(_readLength, exception))
return false;
_readLength = 0;
return true;
} else {
# endif
if (![_delegate respondsToSelector:
@selector(stream:didReadIntoBuffer:length:exception:)])
return false;
if (![_delegate stream: object
didReadIntoBuffer: _buffer
length: _readLength
exception: exception])
return false;
_readLength = 0;
return true;
# ifdef OF_HAVE_BLOCKS
}
# endif
}
# ifdef OF_HAVE_BLOCKS
- (void)dealloc
{
[_block release];
[super dealloc];
}
# endif
@end
@implementation OFRunLoopReadLineQueueItem
- (bool)handleObject: (id)object
{
OFString *line;
id exception = nil;
@try {
line = [object tryReadLineWithEncoding: _encoding];
} @catch (id e) {
line = nil;
exception = e;
}
if (line == nil && ![object isAtEndOfStream] && exception == nil)
return true;
# ifdef OF_HAVE_BLOCKS
if (_block != NULL)
return _block(line, exception);
else {
# endif
if (![_delegate respondsToSelector:
@selector(stream:didReadLine:exception:)])
return false;
return [_delegate stream: object
didReadLine: line
exception: exception];
# ifdef OF_HAVE_BLOCKS
}
# endif
}
# ifdef OF_HAVE_BLOCKS
- (void)dealloc
{
[_block release];
[super dealloc];
}
# endif
@end
@implementation OFRunLoopWriteDataQueueItem
- (bool)handleObject: (id)object
{
size_t length;
id exception = nil;
size_t dataLength = _data.count * _data.itemSize;
OFData *newData, *oldData;
@try {
const char *dataItems = _data.items;
length = [object writeBuffer: dataItems + _writtenLength
length: dataLength - _writtenLength];
} @catch (id e) {
length = 0;
exception = e;
}
_writtenLength += length;
if (_writtenLength != dataLength && exception == nil)
return true;
# ifdef OF_HAVE_BLOCKS
if (_block != NULL) {
newData = _block(_data, _writtenLength, exception);
if (newData == nil)
return false;
oldData = _data;
_data = [newData copy];
[oldData release];
_writtenLength = 0;
return true;
} else {
# endif
if (![_delegate respondsToSelector:
@selector(stream:didWriteData:bytesWritten:exception:)])
return false;
newData = [_delegate stream: object
didWriteData: _data
bytesWritten: _writtenLength
exception: exception];
if (newData == nil)
return false;
oldData = _data;
_data = [newData copy];
[oldData release];
_writtenLength = 0;
return true;
# ifdef OF_HAVE_BLOCKS
}
# endif
}
- (void)dealloc
{
[_data release];
# ifdef OF_HAVE_BLOCKS
[_block release];
# endif
[super dealloc];
}
@end
@implementation OFRunLoopWriteStringQueueItem
- (bool)handleObject: (id)object
{
size_t length;
id exception = nil;
size_t cStringLength = [_string cStringLengthWithEncoding: _encoding];
OFString *newString, *oldString;
@try {
const char *cString = [_string cStringWithEncoding: _encoding];
length = [object writeBuffer: cString + _writtenLength
length: cStringLength - _writtenLength];
} @catch (id e) {
length = 0;
exception = e;
}
_writtenLength += length;
if (_writtenLength != cStringLength && exception == nil)
return true;
# ifdef OF_HAVE_BLOCKS
if (_block != NULL) {
newString = _block(_string, _writtenLength, exception);
if (newString == nil)
return false;
oldString = _string;
_string = [newString copy];
[oldString release];
_writtenLength = 0;
return true;
} else {
# endif
if (![_delegate respondsToSelector: @selector(stream:
didWriteString:encoding:bytesWritten:exception:)])
return false;
newString = [_delegate stream: object
didWriteString: _string
encoding: _encoding
bytesWritten: _writtenLength
exception: exception];
if (newString == nil)
return false;
oldString = _string;
_string = [newString copy];
[oldString release];
_writtenLength = 0;
return true;
# ifdef OF_HAVE_BLOCKS
}
# endif
}
- (void)dealloc
{
[_string release];
# ifdef OF_HAVE_BLOCKS
[_block release];
# endif
[super dealloc];
}
@end
# if !defined(OF_WII) && !defined(OF_NINTENDO_3DS)
@implementation OFRunLoopConnectQueueItem
- (bool)handleObject: (id)object
{
id exception = nil;
int errNo;
if ((errNo = [object of_socketError]) != 0)
exception =
[_delegate of_connectionFailedExceptionForErrNo: errNo];
if ([_delegate respondsToSelector:
@selector(of_socketDidConnect:exception:)]) {
/*
* Make sure we only call the delegate once we removed the
* socket from the kernel event observer. This is necessary as
* otherwise we could try to connect to the next address and it
* would not be re-registered with the kernel event observer,
* which is necessary for some kernel event observers (e.g.
* epoll) even if the fd of the new socket is the same.
*/
OFRunLoop *runLoop = [OFRunLoop currentRunLoop];
OFTimer *timer = [OFTimer
timerWithTimeInterval: 0
target: _delegate
selector: @selector(of_socketDidConnect:
exception:)
object: object
object: exception
repeats: false];
[runLoop addTimer: timer forMode: runLoop.currentMode];
}
return false;
}
@end
# endif
@implementation OFRunLoopAcceptQueueItem
- (bool)handleObject: (id)object
{
id acceptedSocket, exception = nil;
@try {
acceptedSocket = [object accept];
} @catch (id e) {
acceptedSocket = nil;
exception = e;
}
# ifdef OF_HAVE_BLOCKS
if (_block != NULL) {
if ([object isKindOfClass: [OFStreamSocket class]])
return ((of_stream_socket_async_accept_block_t)
_block)(acceptedSocket, exception);
else if ([object isKindOfClass:
[OFSequencedPacketSocket class]])
return
((of_sequenced_packet_socket_async_accept_block_t)
_block)(acceptedSocket, exception);
else
OF_ENSURE(0);
} else {
# endif
if (![_delegate respondsToSelector:
@selector(socket:didAcceptSocket:exception:)])
return false;
return [_delegate socket: object
didAcceptSocket: acceptedSocket
exception: exception];
# ifdef OF_HAVE_BLOCKS
}
# endif
}
# ifdef OF_HAVE_BLOCKS
- (void)dealloc
{
[_block release];
[super dealloc];
}
# endif
@end
@implementation OFRunLoopDatagramReceiveQueueItem
- (bool)handleObject: (id)object
{
size_t length;
of_socket_address_t address;
id exception = nil;
@try {
length = [object receiveIntoBuffer: _buffer
length: _length
sender: &address];
} @catch (id e) {
length = 0;
exception = e;
}
# ifdef OF_HAVE_BLOCKS
if (_block != NULL)
return _block(length, &address, exception);
else {
# endif
if (![_delegate respondsToSelector: @selector(
socket:didReceiveIntoBuffer:length:sender:exception:)])
return false;
return [_delegate socket: object
didReceiveIntoBuffer: _buffer
length: length
sender: &address
exception: exception];
# ifdef OF_HAVE_BLOCKS
}
# endif
}
# ifdef OF_HAVE_BLOCKS
- (void)dealloc
{
[_block release];
[super dealloc];
}
# endif
@end
@implementation OFRunLoopDatagramSendQueueItem
- (bool)handleObject: (id)object
{
id exception = nil;
OFData *newData, *oldData;
@try {
[object sendBuffer: _data.items
length: _data.count * _data.itemSize
receiver: &_receiver];
} @catch (id e) {
exception = e;
}
# ifdef OF_HAVE_BLOCKS
if (_block != NULL) {
newData = _block(_data, &_receiver, exception);
if (newData == nil)
return false;
oldData = _data;
_data = [newData copy];
[oldData release];
return true;
} else {
# endif
if (![_delegate respondsToSelector:
@selector(socket:didSendData:receiver:exception:)])
return false;
newData = [_delegate socket: object
didSendData: _data
receiver: &_receiver
exception: exception];
if (newData == nil)
return false;
oldData = _data;
_data = [newData copy];
[oldData release];
return true;
# ifdef OF_HAVE_BLOCKS
}
# endif
}
- (void)dealloc
{
[_data release];
# ifdef OF_HAVE_BLOCKS
[_block release];
# endif
[super dealloc];
}
@end
@implementation OFRunLoopPacketReceiveQueueItem
- (bool)handleObject: (id)object
{
size_t length;
id exception = nil;
@try {
length = [object receiveIntoBuffer: _buffer length: _length];
} @catch (id e) {
length = 0;
exception = e;
}
# ifdef OF_HAVE_BLOCKS
if (_block != NULL)
return _block(length, exception);
else {
# endif
if (![_delegate respondsToSelector: @selector(
socket:didReceiveIntoBuffer:length:exception:)])
return false;
return [_delegate socket: object
didReceiveIntoBuffer: _buffer
length: length
exception: exception];
# ifdef OF_HAVE_BLOCKS
}
# endif
}
# ifdef OF_HAVE_BLOCKS
- (void)dealloc
{
[_block release];
[super dealloc];
}
# endif
@end
@implementation OFRunLoopPacketSendQueueItem
- (bool)handleObject: (id)object
{
id exception = nil;
OFData *newData, *oldData;
@try {
[object sendBuffer: _data.items
length: _data.count * _data.itemSize];
} @catch (id e) {
exception = e;
}
# ifdef OF_HAVE_BLOCKS
if (_block != NULL) {
newData = _block(_data, exception);
if (newData == nil)
return false;
oldData = _data;
_data = [newData copy];
[oldData release];
return true;
} else {
# endif
if (![_delegate respondsToSelector:
@selector(socket:didSendData:exception:)])
return false;
newData = [_delegate socket: object
didSendData: _data
exception: exception];
if (newData == nil)
return false;
oldData = _data;
_data = [newData copy];
[oldData release];
return true;
# ifdef OF_HAVE_BLOCKS
}
# endif
}
- (void)dealloc
{
[_data release];
# ifdef OF_HAVE_BLOCKS
[_block release];
# endif
[super dealloc];
}
@end
#endif
@implementation OFRunLoop
@synthesize currentMode = _currentMode;
+ (OFRunLoop *)mainRunLoop
{
return mainRunLoop;
}
+ (OFRunLoop *)currentRunLoop
{
#ifdef OF_HAVE_THREADS
return [OFThread currentThread].runLoop;
#else
return [self mainRunLoop];
#endif
}
+ (void)of_setMainRunLoop: (OFRunLoop *)runLoop
{
mainRunLoop = [runLoop retain];
}
static OFRunLoopState *
stateForMode(OFRunLoop *self, of_run_loop_mode_t mode, bool create)
{
OFRunLoopState *state;
#ifdef OF_HAVE_THREADS
[self->_statesMutex lock];
@try {
#endif
state = [self->_states objectForKey: mode];
if (create && state == nil) {
state = [[OFRunLoopState alloc] init];
@try {
[self->_states setObject: state forKey: mode];
} @finally {
[state release];
}
}
#ifdef OF_HAVE_THREADS
} @finally {
[self->_statesMutex unlock];
}
#endif
return state;
}
#ifdef OF_HAVE_SOCKETS
# define NEW_READ(type, object, mode) \
void *pool = objc_autoreleasePoolPush(); \
OFRunLoop *runLoop = [self currentRunLoop]; \
OFRunLoopState *state = stateForMode(runLoop, mode, true); \
OFList *queue = [state->_readQueues objectForKey: object]; \
type *queueItem; \
\
if (queue == nil) { \
queue = [OFList list]; \
[state->_readQueues setObject: queue forKey: object]; \
} \
\
if (queue.count == 0) \
[state->_kernelEventObserver \
addObjectForReading: object]; \
\
queueItem = [[[type alloc] init] autorelease];
# define NEW_WRITE(type, object, mode) \
void *pool = objc_autoreleasePoolPush(); \
OFRunLoop *runLoop = [self currentRunLoop]; \
OFRunLoopState *state = stateForMode(runLoop, mode, true); \
OFList *queue = [state->_writeQueues objectForKey: object]; \
type *queueItem; \
\
if (queue == nil) { \
queue = [OFList list]; \
[state->_writeQueues setObject: queue forKey: object]; \
} \
\
if (queue.count == 0) \
[state->_kernelEventObserver \
addObjectForWriting: object]; \
\
queueItem = [[[type alloc] init] autorelease];
#define QUEUE_ITEM \
[queue appendObject: queueItem]; \
\
objc_autoreleasePoolPop(pool);
+ (void)of_addAsyncReadForStream: (OFStream <OFReadyForReadingObserving> *)
stream
buffer: (void *)buffer
length: (size_t)length
mode: (of_run_loop_mode_t)mode
# ifdef OF_HAVE_BLOCKS
block: (of_stream_async_read_block_t)block
# endif
delegate: (id <OFStreamDelegate>)delegate
{
NEW_READ(OFRunLoopReadQueueItem, stream, mode)
queueItem->_delegate = [delegate retain];
# ifdef OF_HAVE_BLOCKS
queueItem->_block = [block copy];
# endif
queueItem->_buffer = buffer;
queueItem->_length = length;
QUEUE_ITEM
}
+ (void)of_addAsyncReadForStream: (OFStream <OFReadyForReadingObserving> *)
stream
buffer: (void *)buffer
exactLength: (size_t)exactLength
mode: (of_run_loop_mode_t)mode
# ifdef OF_HAVE_BLOCKS
block: (of_stream_async_read_block_t)block
# endif
delegate: (id <OFStreamDelegate>)delegate
{
NEW_READ(OFRunLoopExactReadQueueItem, stream, mode)
queueItem->_delegate = [delegate retain];
# ifdef OF_HAVE_BLOCKS
queueItem->_block = [block copy];
# endif
queueItem->_buffer = buffer;
queueItem->_exactLength = exactLength;
QUEUE_ITEM
}
+ (void)of_addAsyncReadLineForStream: (OFStream <OFReadyForReadingObserving> *)
stream
encoding: (of_string_encoding_t)encoding
mode: (of_run_loop_mode_t)mode
# ifdef OF_HAVE_BLOCKS
block: (of_stream_async_read_line_block_t)block
# endif
delegate: (id <OFStreamDelegate>)delegate
{
NEW_READ(OFRunLoopReadLineQueueItem, stream, mode)
queueItem->_delegate = [delegate retain];
# ifdef OF_HAVE_BLOCKS
queueItem->_block = [block copy];
# endif
queueItem->_encoding = encoding;
QUEUE_ITEM
}
+ (void)of_addAsyncWriteForStream: (OFStream <OFReadyForWritingObserving> *)
stream
data: (OFData *)data
mode: (of_run_loop_mode_t)mode
# ifdef OF_HAVE_BLOCKS
block: (of_stream_async_write_data_block_t)block
# endif
delegate: (id <OFStreamDelegate>)delegate
{
NEW_WRITE(OFRunLoopWriteDataQueueItem, stream, mode)
queueItem->_delegate = [delegate retain];
# ifdef OF_HAVE_BLOCKS
queueItem->_block = [block copy];
# endif
queueItem->_data = [data copy];
QUEUE_ITEM
}
+ (void)of_addAsyncWriteForStream: (OFStream <OFReadyForWritingObserving> *)
stream
string: (OFString *)string
encoding: (of_string_encoding_t)encoding
mode: (of_run_loop_mode_t)mode
# ifdef OF_HAVE_BLOCKS
block: (of_stream_async_write_string_block_t)block
# endif
delegate: (id <OFStreamDelegate>)delegate
{
NEW_WRITE(OFRunLoopWriteStringQueueItem, stream, mode)
queueItem->_delegate = [delegate retain];
# ifdef OF_HAVE_BLOCKS
queueItem->_block = [block copy];
# endif
queueItem->_string = [string copy];
queueItem->_encoding = encoding;
QUEUE_ITEM
}
# if !defined(OF_WII) && !defined(OF_NINTENDO_3DS)
+ (void)of_addAsyncConnectForSocket: (id)sock
mode: (of_run_loop_mode_t)mode
delegate: (id <OFRunLoopConnectDelegate>)delegate
{
NEW_WRITE(OFRunLoopConnectQueueItem, sock, mode)
queueItem->_delegate = [delegate retain];
QUEUE_ITEM
}
# endif
+ (void)of_addAsyncAcceptForSocket: (id)sock
mode: (of_run_loop_mode_t)mode
block: (id)block
delegate: (id)delegate
{
NEW_READ(OFRunLoopAcceptQueueItem, sock, mode)
queueItem->_delegate = [delegate retain];
# ifdef OF_HAVE_BLOCKS
queueItem->_block = [block copy];
# endif
QUEUE_ITEM
}
+ (void)of_addAsyncReceiveForDatagramSocket: (OFDatagramSocket *)sock
buffer: (void *)buffer
length: (size_t)length
mode: (of_run_loop_mode_t)mode
# ifdef OF_HAVE_BLOCKS
block: (of_datagram_socket_async_receive_block_t)block
# endif
delegate: (id <OFDatagramSocketDelegate>)delegate
{
NEW_READ(OFRunLoopDatagramReceiveQueueItem, sock, mode)
queueItem->_delegate = [delegate retain];
# ifdef OF_HAVE_BLOCKS
queueItem->_block = [block copy];
# endif
queueItem->_buffer = buffer;
queueItem->_length = length;
QUEUE_ITEM
}
+ (void)of_addAsyncSendForDatagramSocket: (OFDatagramSocket *)sock
data: (OFData *)data
receiver: (const of_socket_address_t *)receiver
mode: (of_run_loop_mode_t)mode
# ifdef OF_HAVE_BLOCKS
block: (of_datagram_socket_async_send_data_block_t)block
# endif
delegate: (id <OFDatagramSocketDelegate>)delegate
{
NEW_WRITE(OFRunLoopDatagramSendQueueItem, sock, mode)
queueItem->_delegate = [delegate retain];
# ifdef OF_HAVE_BLOCKS
queueItem->_block = [block copy];
# endif
queueItem->_data = [data copy];
queueItem->_receiver = *receiver;
QUEUE_ITEM
}
+ (void)of_addAsyncReceiveForSequencedPacketSocket: (OFSequencedPacketSocket *)
sock
buffer: (void *)buffer
length: (size_t)length
mode: (of_run_loop_mode_t)mode
# ifdef OF_HAVE_BLOCKS
block: (of_sequenced_packet_socket_async_receive_block_t)block
# endif
delegate: (id <OFSequencedPacketSocketDelegate>)delegate
{
NEW_READ(OFRunLoopPacketReceiveQueueItem, sock, mode)
queueItem->_delegate = [delegate retain];
# ifdef OF_HAVE_BLOCKS
queueItem->_block = [block copy];
# endif
queueItem->_buffer = buffer;
queueItem->_length = length;
QUEUE_ITEM
}
+ (void)of_addAsyncSendForSequencedPacketSocket: (OFSequencedPacketSocket *)sock
data: (OFData *)data
mode: (of_run_loop_mode_t)mode
# ifdef OF_HAVE_BLOCKS
block: (of_sequenced_packet_socket_async_send_data_block_t)block
# endif
delegate: (id <OFSequencedPacketSocketDelegate>)delegate
{
NEW_WRITE(OFRunLoopPacketSendQueueItem, sock, mode)
queueItem->_delegate = [delegate retain];
# ifdef OF_HAVE_BLOCKS
queueItem->_block = [block copy];
# endif
queueItem->_data = [data copy];
QUEUE_ITEM
}
# undef NEW_READ
# undef NEW_WRITE
# undef QUEUE_ITEM
+ (void)of_cancelAsyncRequestsForObject: (id)object
mode: (of_run_loop_mode_t)mode
{
void *pool = objc_autoreleasePoolPush();
OFRunLoop *runLoop = [self currentRunLoop];
OFRunLoopState *state = stateForMode(runLoop, mode, false);
OFList *queue;
if (state == nil)
return;
if ((queue = [state->_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];
[state->_kernelEventObserver removeObjectForWriting: object];
[state->_writeQueues removeObjectForKey: object];
}
if ((queue = [state->_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.
*/
[queue removeAllObjects];
[state->_kernelEventObserver removeObjectForReading: object];
[state->_readQueues removeObjectForKey: object];
}
objc_autoreleasePoolPop(pool);
}
#endif
- (instancetype)init
{
self = [super init];
@try {
OFRunLoopState *state;
_states = [[OFMutableDictionary alloc] init];
state = [[OFRunLoopState alloc] init];
@try {
[_states setObject: state
forKey: of_run_loop_mode_default];
} @finally {
[state release];
}
#ifdef OF_HAVE_THREADS
_statesMutex = [[OFMutex alloc] init];
#endif
} @catch (id e) {
[self release];
@throw e;
}
return self;
}
- (void)dealloc
{
[_states release];
#ifdef OF_HAVE_THREADS
[_statesMutex release];
#endif
[super dealloc];
}
- (void)addTimer: (OFTimer *)timer
{
[self addTimer: timer forMode: of_run_loop_mode_default];
}
- (void)addTimer: (OFTimer *)timer forMode: (of_run_loop_mode_t)mode
{
OFRunLoopState *state = stateForMode(self, mode, true);
#ifdef OF_HAVE_THREADS
[state->_timersQueueMutex lock];
@try {
#endif
[state->_timersQueue insertObject: timer];
#ifdef OF_HAVE_THREADS
} @finally {
[state->_timersQueueMutex unlock];
}
#endif
[timer of_setInRunLoop: self mode: mode];
#if defined(OF_HAVE_SOCKETS)
[state->_kernelEventObserver cancel];
#elif defined(OF_HAVE_THREADS)
[state->_condition signal];
#endif
}
- (void)of_removeTimer: (OFTimer *)timer forMode: (of_run_loop_mode_t)mode
{
OFRunLoopState *state = stateForMode(self, mode, false);
if (state == nil)
return;
#ifdef OF_HAVE_THREADS
[state->_timersQueueMutex lock];
@try {
#endif
of_list_object_t *iter;
for (iter = state->_timersQueue.firstListObject; iter != NULL;
iter = iter->next) {
if ([iter->object isEqual: timer]) {
[state->_timersQueue removeListObject: iter];
break;
}
}
#ifdef OF_HAVE_THREADS
} @finally {
[state->_timersQueueMutex unlock];
}
#endif
}
#ifdef OF_AMIGAOS
- (void)addExecSignal: (ULONG)signal target: (id)target selector: (SEL)selector
{
[self addExecSignal: signal
forMode: of_run_loop_mode_default
target: target
selector: selector];
}
- (void)addExecSignal: (ULONG)signal
forMode: (of_run_loop_mode_t)mode
target: (id)target
selector: (SEL)selector
{
OFRunLoopState *state = stateForMode(self, mode, true);
# ifdef OF_HAVE_THREADS
[state->_execSignalsMutex lock];
@try {
# endif
[state->_execSignals addItem: &signal];
[state->_execSignalsTargets addObject: target];
[state->_execSignalsSelectors addItem: &selector];
# ifdef OF_HAVE_SOCKETS
state->_kernelEventObserver.execSignalMask |= (1ul << signal);
# elif defined(OF_HAVE_THREADS)
state->_execSignalMask |= (1ul << signal);
# endif
# ifdef OF_HAVE_THREADS
} @finally {
[state->_execSignalsMutex unlock];
}
# endif
# if defined(OF_HAVE_SOCKETS)
[state->_kernelEventObserver cancel];
# elif defined(OF_HAVE_THREADS)
[state->_condition signal];
# endif
}
- (void)removeExecSignal: (ULONG)signal
target: (id)target
selector: (SEL)selector
{
[self removeExecSignal: signal
forMode: of_run_loop_mode_default
target: target
selector: selector];
}
- (void)removeExecSignal: (ULONG)signal
forMode: (of_run_loop_mode_t)mode
target: (id)target
selector: (SEL)selector
{
OFRunLoopState *state = stateForMode(self, mode, false);
if (state == nil)
return;
# ifdef OF_HAVE_THREADS
[state->_execSignalsMutex lock];
@try {
# endif
const ULONG *signals = state->_execSignals.items;
const id *targets = state->_execSignalsTargets.objects;
const SEL *selectors = state->_execSignalsSelectors.items;
size_t count = state->_execSignals.count;
bool found = false;
ULONG newMask = 0;
for (size_t i = 0; i < count; i++) {
if (!found && signals[i] == signal &&
targets[i] == target && selectors[i] == selector) {
[state->_execSignals removeItemAtIndex: i];
[state->_execSignalsTargets
removeObjectAtIndex: i];
[state->_execSignalsSelectors
removeItemAtIndex: i];
found = true;
} else
newMask |= (1ul << signals[i]);
}
# ifdef OF_HAVE_SOCKETS
state->_kernelEventObserver.execSignalMask = newMask;
# elif defined(OF_HAVE_THREADS)
state->_execSignalMask = newMask;
# endif
# ifdef OF_HAVE_THREADS
} @finally {
[state->_execSignalsMutex unlock];
}
# endif
# if defined(OF_HAVE_SOCKETS)
[state->_kernelEventObserver cancel];
# elif defined(OF_HAVE_THREADS)
[state->_condition signal];
# endif
}
#endif
- (void)run
{
[self runUntilDate: nil];
}
- (void)runUntilDate: (OFDate *)deadline
{
_stop = false;
while (!_stop &&
(deadline == nil || deadline.timeIntervalSinceNow >= 0))
[self runMode: of_run_loop_mode_default beforeDate: deadline];
}
- (void)runMode: (of_run_loop_mode_t)mode beforeDate: (OFDate *)deadline
{
void *pool = objc_autoreleasePoolPush();
of_run_loop_mode_t previousMode = _currentMode;
OFRunLoopState *state = stateForMode(self, mode, false);
if (state == nil)
return;
_currentMode = mode;
@try {
OFDate *nextTimer;
#if defined(OF_AMIGAOS) && !defined(OF_HAVE_SOCKETS) && defined(OF_HAVE_THREADS)
ULONG signalMask;
#endif
for (;;) {
OFTimer *timer;
#ifdef OF_HAVE_THREADS
[state->_timersQueueMutex lock];
@try {
#endif
of_list_object_t *listObject =
state->_timersQueue.firstListObject;
if (listObject != NULL && [listObject->object
fireDate].timeIntervalSinceNow <= 0) {
timer = [[listObject->object
retain] autorelease];
[state->_timersQueue
removeListObject: listObject];
[timer of_setInRunLoop: nil mode: nil];
} else
break;
#ifdef OF_HAVE_THREADS
} @finally {
[state->_timersQueueMutex unlock];
}
#endif
if (timer.valid) {
[timer fire];
return;
}
}
#ifdef OF_HAVE_THREADS
[state->_timersQueueMutex lock];
@try {
#endif
nextTimer = [[state->_timersQueue
firstObject] fireDate];
#ifdef OF_HAVE_THREADS
} @finally {
[state->_timersQueueMutex unlock];
}
#endif
/* Watch for I/O events until the next timer is due */
if (nextTimer != nil || deadline != nil) {
OFTimeInterval timeout;
if (nextTimer != nil && deadline == nil)
timeout = nextTimer.timeIntervalSinceNow;
else if (nextTimer == nil && deadline != nil)
timeout = deadline.timeIntervalSinceNow;
else
timeout = [nextTimer earlierDate: deadline]
.timeIntervalSinceNow;
if (timeout < 0)
timeout = 0;
#if defined(OF_HAVE_SOCKETS)
@try {
[state->_kernelEventObserver
observeForTimeInterval: timeout];
} @catch (OFObserveFailedException *e) {
if (e.errNo != EINTR)
@throw e;
}
#elif defined(OF_HAVE_THREADS)
[state->_condition lock];
# ifdef OF_AMIGAOS
signalMask = state->_execSignalMask;
[state->_condition waitForTimeInterval: timeout
orExecSignal: &signalMask];
if (signalMask != 0)
[state execSignalWasReceived: signalMask];
# else
[state->_condition waitForTimeInterval: timeout];
# endif
[state->_condition unlock];
#else
[OFThread sleepForTimeInterval: timeout];
#endif
} else {
/*
* No more timers and no deadline: Just watch for I/O
* until we get an event. If a timer is added by
* another thread, it cancels the observe.
*/
#if defined(OF_HAVE_SOCKETS)
@try {
[state->_kernelEventObserver observe];
} @catch (OFObserveFailedException *e) {
if (e.errNo != EINTR)
@throw e;
}
#elif defined(OF_HAVE_THREADS)
[state->_condition lock];
# ifdef OF_AMIGAOS
signalMask = state->_execSignalMask;
[state->_condition
waitForConditionOrExecSignal: &signalMask];
if (signalMask != 0)
[state execSignalWasReceived: signalMask];
# else
[state->_condition wait];
# endif
[state->_condition unlock];
#else
[OFThread sleepForTimeInterval: 86400];
#endif
}
objc_autoreleasePoolPop(pool);
} @finally {
_currentMode = previousMode;
}
}
- (void)stop
{
OFRunLoopState *state =
stateForMode(self, of_run_loop_mode_default, false);
_stop = true;
if (state == nil)
return;
#if defined(OF_HAVE_SOCKETS)
[state->_kernelEventObserver cancel];
#elif defined(OF_HAVE_THREADS)
[state->_condition signal];
#endif
}
@end