ObjFW  Diff

# import "OFKernelEventObserver_poll.h"
#endif
#if defined(HAVE_SYS_SELECT_H) || defined(_WIN32)
# import "OFKernelEventObserver_select.h"
#endif

#import "OFInitializationFailedException.h"
#import "OFInvalidArgumentException.h"
#import "OFOutOfRangeException.h"

#import "autorelease.h"
#import "macros.h"
#import "socket_helpers.h"

enum {

#ifndef OF_HAVE_PIPE
		struct sockaddr_in cancelAddr2;
# ifndef __wii__
		socklen_t cancelAddrLen;
# endif
#endif

		_readStreams = [[OFMutableArray alloc] init];
		_writeStreams = [[OFMutableArray alloc] init];
		_readObjects = [[OFMutableArray alloc] init];
		_writeObjects = [[OFMutableArray alloc] init];
		_queue = [[OFMutableArray alloc] init];
		_queueInfo = [[OFDataArray alloc]
		    initWithItemSize: sizeof(int)];
		_queueFDs = [[OFDataArray alloc] initWithItemSize: sizeof(int)];

#ifdef OF_HAVE_PIPE
		if (pipe(_cancelFD))

		    &cancelAddrLen))
			@throw [OFInitializationFailedException
			    exceptionWithClass: [self class]];
# endif
#endif

		_maxFD = _cancelFD[0];
		_FDToStream = [self allocMemoryWithSize: sizeof(OFStream*)
		_FDToObject = [self allocMemoryWithSize: sizeof(id)
						  count: _maxFD + 1];
		_FDToStream[_cancelFD[0]] = nil;
		_FDToObject[_cancelFD[0]] = nil;

#ifdef OF_HAVE_THREADS
		_mutex = [[OFMutex alloc] init];
#endif
	} @catch (id e) {
		[self release];
		@throw e;
	}

	return self;
}

- (void)dealloc
{
	close(_cancelFD[0]);
	close(_cancelFD[1]);

	[_readStreams release];
	[_writeStreams release];
	[_readObjects release];
	[_writeObjects release];
	[_queue release];
	[_queueInfo release];
	[_queueFDs release];
#ifdef OF_HAVE_THREADS
	[_mutex release];
#endif

	[super dealloc];
}

- (id <OFKernelEventObserverDelegate>)delegate
{
	return _delegate;
}

- (void)setDelegate: (id <OFKernelEventObserverDelegate>)delegate
{
	_delegate = delegate;
}

- (void)addStreamForReading: (OFStream*)stream
- (void)addObjectForReading: (id <OFReadyForReadingObserving>)object
{
#ifdef OF_HAVE_THREADS
	[_mutex lock];
#endif
	@try {
		int qi = QUEUE_ADD | QUEUE_READ;
		int fd = [stream fileDescriptorForReading];
		int fd = [object fileDescriptorForReading];

		[_queue addObject: stream];
		[_queue addObject: object];
		[_queueInfo addItem: &qi];
		[_queueFDs addItem: &fd];
	} @finally {
#ifdef OF_HAVE_THREADS
		[_mutex unlock];
#endif
	}

	[self cancel];
}

- (void)addObjectForWriting: (id <OFReadyForWritingObserving>)object
{
#ifdef OF_HAVE_THREADS
	[_mutex lock];
#endif
	@try {
		int qi = QUEUE_ADD | QUEUE_WRITE;
		int fd = [object fileDescriptorForWriting];

		[_queue addObject: object];
		[_queueInfo addItem: &qi];
		[_queueFDs addItem: &fd];
	} @finally {
#ifdef OF_HAVE_THREADS
		[_mutex unlock];
#endif
	}

	[self cancel];
}

- (void)addStreamForWriting: (OFStream*)stream
- (void)removeObjectForReading: (id <OFReadyForReadingObserving>)object
{
#ifdef OF_HAVE_THREADS
	[_mutex lock];
#endif
	@try {
		int qi = QUEUE_ADD | QUEUE_WRITE;
		int fd = [stream fileDescriptorForWriting];
		int qi = QUEUE_REMOVE | QUEUE_READ;
		int fd = [object fileDescriptorForReading];

		[_queue addObject: stream];
		[_queue addObject: object];
		[_queueInfo addItem: &qi];
		[_queueFDs addItem: &fd];
	} @finally {
#ifdef OF_HAVE_THREADS
		[_mutex unlock];
#endif
	}

	[self cancel];
}

- (void)removeStreamForReading: (OFStream*)stream
{
#ifdef OF_HAVE_THREADS
	[_mutex lock];
#endif
	@try {
		int qi = QUEUE_REMOVE | QUEUE_READ;
		int fd = [stream fileDescriptorForReading];

- (void)removeObjectForWriting: (id <OFReadyForWritingObserving>)object
		[_queue addObject: stream];
		[_queueInfo addItem: &qi];
		[_queueFDs addItem: &fd];
	} @finally {
#ifdef OF_HAVE_THREADS
		[_mutex unlock];
#endif
	}

	[self cancel];
}

- (void)removeStreamForWriting: (OFStream*)stream
{
#ifdef OF_HAVE_THREADS
	[_mutex lock];
#endif
	@try {
		int qi = QUEUE_REMOVE | QUEUE_WRITE;
		int fd = [stream fileDescriptorForWriting];
		int fd = [object fileDescriptorForWriting];

		[_queue addObject: stream];
		[_queue addObject: object];
		[_queueInfo addItem: &qi];
		[_queueFDs addItem: &fd];
	} @finally {
#ifdef OF_HAVE_THREADS
		[_mutex unlock];
#endif
	}

- (void)OF_processQueue
{
#ifdef OF_HAVE_THREADS
	[_mutex lock];
#endif
	@try {
		OFStream **queueObjects = [_queue objects];
		id *queueObjects = [_queue objects];
		int *queueInfoItems = [_queueInfo items];
		int *queueFDsItems = [_queueFDs items];
		size_t i, count = [_queue count];

		for (i = 0; i < count; i++) {
			OFStream *stream = queueObjects[i];
			id object = queueObjects[i];
			int action = queueInfoItems[i];
			int fd = queueFDsItems[i];

			if ((action & QUEUE_ACTION) == QUEUE_ADD) {
				if (fd > _maxFD) {
					_maxFD = fd;
					_FDToStream = [self
					    resizeMemory: _FDToStream
						    size: sizeof(OFStream*)
					_FDToObject = [self
					    resizeMemory: _FDToObject
						    size: sizeof(id)
						   count: _maxFD + 1];
				}

				_FDToStream[fd] = stream;
				_FDToObject[fd] = object;
			}

			if ((action & QUEUE_ACTION) == QUEUE_REMOVE) {
				/* FIXME: Maybe downsize? */
				_FDToStream[fd] = nil;
				_FDToObject[fd] = nil;
			}

			switch (action) {
			case QUEUE_ADD | QUEUE_READ:
				[_readStreams addObject: stream];
				[_readObjects addObject: object];

				[self OF_addFileDescriptorForReading: fd];

				break;
			case QUEUE_ADD | QUEUE_WRITE:
				[_writeStreams addObject: stream];
				[_writeObjects addObject: object];

				[self OF_addFileDescriptorForWriting: fd];

				break;
			case QUEUE_REMOVE | QUEUE_READ:
				[_readStreams removeObjectIdenticalTo: stream];
				[_readObjects removeObjectIdenticalTo: object];

				[self OF_removeFileDescriptorForReading: fd];

				break;
			case QUEUE_REMOVE | QUEUE_WRITE:
				[_writeStreams removeObjectIdenticalTo: stream];
				[_writeObjects removeObjectIdenticalTo: object];

				[self OF_removeFileDescriptorForWriting: fd];

				break;
			default:
				assert(0);
			}

	OF_ENSURE(sendto(_cancelFD[1], "", 1, 0, (struct sockaddr*)&_cancelAddr,
	    sizeof(_cancelAddr)) > 0);
#endif
}

- (bool)OF_processCache
{
	OFStream **objects = [_readStreams objects];
	size_t i, count = [_readStreams count];
	id *objects = [_readObjects objects];
	size_t i, count = [_readObjects count];
	bool foundInCache = false;

	for (i = 0; i < count; i++) {
		if ([objects[i] isKindOfClass: [OFStream class]] &&
		if ([objects[i] numberOfBytesInReadBuffer] > 0 &&
		    [objects[i] numberOfBytesInReadBuffer] > 0 &&
		    ![objects[i] OF_isWaitingForDelimiter]) {
			void *pool = objc_autoreleasePoolPush();

			if ([_delegate respondsToSelector:
			    @selector(streamIsReadyForReading:)])
				[_delegate streamIsReadyForReading: objects[i]];
			    @selector(objectsIsReadyForReading:)])
				[_delegate objectIsReadyForReading: objects[i]];

			foundInCache = true;

			objc_autoreleasePoolPop(pool);
		}
	}