ObjFW  Check-in [220513a3f5]

@interface OFKernelEventObserver (OF_PRIVATE_CATEGORY)
- (void)OF_addObjectForReading: (id)object;
- (void)OF_addObjectForWriting: (id)object;
- (void)OF_removeObjectForReading: (id)object;
- (void)OF_removeObjectForWriting: (id)object;
- (void)OF_processQueueAndStoreRemovedIn: (OFMutableArray*)removed;
- (void)OF_processReadBuffers;
@end

- (void)observe;

/*!
 * @brief Observes all objects until an event happens on an object or the
 *	  timeout is reached.
 *
 * @param timeInterval The time to wait for an event, in seconds

 */
- (void)observeForTimeInterval: (of_time_interval_t)timeInterval;

/*!
 * @brief Observes all objects until an event happens on an object or the
 *	  specified date is reached.
 *
 * @param date The until which to observe

 */
- (void)observeUntilDate: (OFDate*)date;

/*!
 * @brief Cancels the currently blocking observe call.
 *
 * This is automatically done when a new object is added or removed by another
 * thread, but in some circumstances, it might be desirable for a thread to
 * manually stop the observe running in another thread.

}

- (void)observe
{
	[self observeForTimeInterval: -1];
}

- (void)observeForTimeInterval: (of_time_interval_t)timeInterval
{
	OF_UNRECOGNIZED_SELECTOR
}

- (void)observeUntilDate: (OFDate*)date
{
	[self observeForTimeInterval: [date timeIntervalSinceNow]];
}

- (void)cancel
{
#ifdef OF_HAVE_PIPE
	OF_ENSURE(write(_cancelFD[1], "", 1) > 0);
#else
	OF_ENSURE(sendto(_cancelFD[1], "", 1, 0, (struct sockaddr*)&_cancelAddr,
	    sizeof(_cancelAddr)) > 0);
#endif
}

- (void)OF_processReadBuffers
{
	id const *objects = [_readObjects objects];
	size_t i, count = [_readObjects count];


	for (i = 0; i < count; i++) {
		void *pool = objc_autoreleasePoolPush();

		if ([objects[i] isKindOfClass: [OFStream class]] &&
		    [objects[i] hasDataInReadBuffer] &&
		    ![objects[i] OF_isWaitingForDelimiter] &&
		    [_delegate respondsToSelector:
		    @selector(objectIsReadyForReading:)])
			[_delegate objectIsReadyForReading: objects[i]];




		objc_autoreleasePoolPop(pool);
	}
}









@end

- (void)OF_removeObjectForWriting: (id)object
{
	[self OF_removeObject: object
	       fileDescriptor: [object fileDescriptorForWriting]
		       events: EPOLLOUT];
}

- (void)observeForTimeInterval: (of_time_interval_t)timeInterval
{
	OFNull *nullObject = [OFNull null];
	void *pool = objc_autoreleasePoolPush();
	struct epoll_event eventList[EVENTLIST_SIZE];
	int i, events;

	[self OF_processQueueAndStoreRemovedIn: nil];

	[self OF_processReadBuffers];




	objc_autoreleasePoolPop(pool);

	events = epoll_wait(_epfd, eventList, EVENTLIST_SIZE,
	    (timeInterval != -1 ? timeInterval * 1000 : -1));

	if (events < 0)
		return [OFObserveFailedException exceptionWithObserver: self
								 errNo: errno];




	for (i = 0; i < events; i++) {
		if (eventList[i].data.ptr == nullObject) {
			char buffer;

			assert(eventList[i].events == EPOLLIN);
			OF_ENSURE(read(_cancelFD[0], &buffer, 1) == 1);

			continue;
		}

		if (eventList[i].events & EPOLLIN) {
			pool = objc_autoreleasePoolPush();

			if ([_delegate respondsToSelector:
			    @selector(objectIsReadyForReading:)])
				[_delegate objectIsReadyForReading:
				    eventList[i].data.ptr];



			objc_autoreleasePoolPop(pool);
		}

		if (eventList[i].events & EPOLLOUT) {
			pool = objc_autoreleasePoolPush();

			if ([_delegate respondsToSelector:
			    @selector(objectIsReadyForWriting:)])
				[_delegate objectIsReadyForWriting:
				    eventList[i].data.ptr];



			objc_autoreleasePoolPop(pool);
		}

		assert((eventList[i].events & ~(EPOLLIN | EPOLLOUT)) == 0);
	}
}





@end

	memset(&event, 0, sizeof(event));
	event.ident = [object fileDescriptorForWriting];
	event.filter = EVFILT_WRITE;
	event.flags = EV_DELETE;
	[_changeList addItem: &event];
}

- (void)observeForTimeInterval: (of_time_interval_t)timeInterval
{
	void *pool = objc_autoreleasePoolPush();
	struct timespec timeout;
	struct kevent eventList[EVENTLIST_SIZE];
	int i, events;

	timeout.tv_sec = (time_t)timeInterval;
	timeout.tv_nsec = lrint((timeInterval - timeout.tv_sec) * 1000000000);

	/*
	 * Make sure to keep the streams retained and thus the file descriptors
	 * valid until the actual change has been performed.
	 */
	[self OF_processQueueAndStoreRemovedIn: _removedArray];

	[self OF_processReadBuffers];




	objc_autoreleasePoolPop(pool);

	events = kevent(_kernelQueue, [_changeList items],
	    (int)[_changeList count], eventList, EVENTLIST_SIZE,
	    (timeInterval != -1 ? &timeout : NULL));

	if (events < 0)
		@throw [OFObserveFailedException exceptionWithObserver: self
								 errNo: errno];

	[_changeList removeAllItems];
	[_removedArray removeAllObjects];




	for (i = 0; i < events; i++) {
		if (eventList[i].flags & EV_ERROR)
			@throw [OFObserveFailedException
			    exceptionWithObserver: self
					    errNo: (int)eventList[i].data];

		if (eventList[i].ident == _cancelFD[0]) {

				    (id)eventList[i].udata];
			break;
		default:
			assert(0);
		}

		objc_autoreleasePoolPop(pool);
	}

}






@end

- (void)OF_removeObjectForWriting: (id)object
{
	[self OF_removeObject: object
	       fileDescriptor: [object fileDescriptorForWriting]
		       events: POLLOUT];
}

- (void)observeForTimeInterval: (of_time_interval_t)timeInterval
{
	void *pool = objc_autoreleasePoolPush();
	struct pollfd *FDs;
	int events;
	size_t i, nFDs;

	[self OF_processQueueAndStoreRemovedIn: nil];

	[self OF_processReadBuffers];




	objc_autoreleasePoolPop(pool);

	FDs = [_FDs items];
	nFDs = [_FDs count];

#ifdef OPEN_MAX
	if (nFDs > OPEN_MAX)
		@throw [OFOutOfRangeException exception];
#endif

	events = poll(FDs, (nfds_t)nFDs,
	    (int)(timeInterval != -1 ? timeInterval * 1000 : -1));

	if (events < 0)
		@throw [OFObserveFailedException exceptionWithObserver: self
								 errNo: errno];




	for (i = 0; i < nFDs; i++) {
		if (FDs[i].fd > _maxFD)
			@throw [OFOutOfRangeException exception];

		if (FDs[i].revents & POLLIN) {
			if (FDs[i].fd == _cancelFD[0]) {
				char buffer;

			if ([_delegate respondsToSelector:
			    @selector(objectIsReadyForReading:)])
				[_delegate objectIsReadyForReading:
				    _FDToObject[FDs[i].fd]];

			objc_autoreleasePoolPop(pool);


		}

		if (FDs[i].revents & POLLOUT) {
			pool = objc_autoreleasePoolPush();

			if ([_delegate respondsToSelector:
			    @selector(objectIsReadyForWriting:)])
				[_delegate objectIsReadyForWriting:
				    _FDToObject[FDs[i].fd]];

			objc_autoreleasePoolPop(pool);


		}

		FDs[i].revents = 0;
	}
}





@end

	if (fd >= FD_SETSIZE)
		@throw [OFOutOfRangeException exception];

	FD_CLR(fd, &_writeFDs);
}

- (void)observeForTimeInterval: (of_time_interval_t)timeInterval
{
	void *pool = objc_autoreleasePoolPush();
	id const *objects;
	fd_set readFDs;
	fd_set writeFDs;
	struct timeval timeout;
	int events;
	size_t i, count;

	[self OF_processQueueAndStoreRemovedIn: nil];

	[self OF_processReadBuffers];




	objc_autoreleasePoolPop(pool);

#ifdef FD_COPY
	FD_COPY(&_readFDs, &readFDs);
	FD_COPY(&_writeFDs, &writeFDs);
#else

	events = select(_maxFD + 1, &readFDs, &writeFDs, NULL,
	    (timeInterval != -1 ? &timeout : NULL));

	if (events < 0)
		@throw [OFObserveFailedException exceptionWithObserver: self
								 errNo: errno];




	if (FD_ISSET(_cancelFD[0], &readFDs)) {
		char buffer;

#ifndef _WIN32
		OF_ENSURE(read(_cancelFD[0], &buffer, 1) == 1);
#else
		OF_ENSURE(recvfrom(_cancelFD[0], &buffer, 1, 0, NULL,
		    NULL) == 1);
#endif
	}

	objects = [_readObjects objects];
	count = [_readObjects count];

	for (i = 0; i < count; i++) {
		int fd = [objects[i] fileDescriptorForReading];

		pool = objc_autoreleasePoolPush();

		if (FD_ISSET(fd, &readFDs) && [_delegate respondsToSelector:

		    @selector(objectIsReadyForReading:)])
			[_delegate objectIsReadyForReading: objects[i]];




		objc_autoreleasePoolPop(pool);
	}

	objects = [_writeObjects objects];
	count = [_writeObjects count];

	for (i = 0; i < count; i++) {
		int fd = [objects[i] fileDescriptorForWriting];

		pool = objc_autoreleasePoolPush();

		if (FD_ISSET(fd, &writeFDs) && [_delegate respondsToSelector:

		    @selector(objectIsReadyForWriting:)])
			[_delegate objectIsReadyForWriting: objects[i]];




		objc_autoreleasePoolPop(pool);
	}
}





@end

 * file.
 */

#include "config.h"

#import "OFKernelEventObserver.h"
#import "OFString.h"
#import "OFDate.h"
#import "OFTCPSocket.h"
#import "OFAutoreleasePool.h"

#if defined(HAVE_SYS_SELECT_H) || defined(_WIN32)
# import "OFKernelEventObserver_select.h"
#endif
#if defined(HAVE_POLL_H) || defined(__wii__)
# import "OFKernelEventObserver_poll.h"
#endif
#ifdef HAVE_EPOLL
# import "OFKernelEventObserver_epoll.h"
#endif
#ifdef HAVE_KQUEUE
# import "OFKernelEventObserver_kqueue.h"
#endif

#import "TestsAppDelegate.h"

#define EXPECTED_EVENTS 3

static OFString *module;

@interface ObserverTest: OFObject
{
@public
	TestsAppDelegate *_testsAppDelegate;
	OFKernelEventObserver *_observer;
	OFTCPSocket *_server, *_client, *_accepted;
	size_t _events;



}

- (void)run;
@end

@implementation ObserverTest
- initWithTestsAppDelegate: (TestsAppDelegate*)testsAppDelegate
{
	self = [super init];

	@try {
		uint16_t port;

		_testsAppDelegate = testsAppDelegate;

		_server = [[OFTCPSocket alloc] init];
		port = [_server bindToHost: @"127.0.0.1"
				      port: 0];
		[_server listen];

		_client = [[OFTCPSocket alloc] init];
		[_client connectToHost: @"127.0.0.1"
				  port: port];

		[_client writeBuffer: "0"
			      length: 1];
	} @catch (id e) {
		[self release];
		@throw e;
	}

	return self;
}

- (void)dealloc
{
	[_server release];
	[_client release];
	[_accepted release];

	[super dealloc];
}

- (void)run
{
	OFDate *deadline;
	bool deadlineExceeded = false;

	[_testsAppDelegate outputTesting: @"-[observe] with listening socket"
				inModule: module];

	deadline = [OFDate dateWithTimeIntervalSinceNow: 1];
	while (_events < EXPECTED_EVENTS) {
		if ([deadline timeIntervalSinceNow] < 0) {
			deadlineExceeded = true;
			break;
		}

		[_observer observeForTimeInterval: 0.01];
	}

	if (!deadlineExceeded)
		[_testsAppDelegate
		    outputSuccess: @"-[observe] not exceeding deadline"
			 inModule: module];
	else
		[_testsAppDelegate
		    outputFailure: @"-[observe] not exceeding deadline"
			 inModule: module];

	if (_events == EXPECTED_EVENTS)
		[_testsAppDelegate
		    outputSuccess: @"-[observe] handling all events"
			 inModule: module];
	else
		[_testsAppDelegate
		    outputFailure: @"-[observe] handling all events"
			 inModule: module];
}

- (void)objectIsReadyForReading: (id)object
{

	char buf;

	switch (_events++) {
	case 0:
		if (object == _server)
			[_testsAppDelegate
			    outputSuccess: @"-[observe] with listening socket"
				 inModule: module];
		else
			[_testsAppDelegate
			    outputFailure: @"-[observe] with listening socket"
				 inModule: module];


		_accepted = [[object accept] retain];
		[_observer addObjectForReading: _accepted];

		[_testsAppDelegate
		    outputTesting: @"-[observe] with data to read available"
			 inModule: module];

		break;
	case 1:
		if (object == _accepted &&
		    [object readIntoBuffer: &buf
				    length: 1] == 1 && buf == '0')
			[_testsAppDelegate
			    outputSuccess: @"-[observe] with data to read "
					   @"available"
				 inModule: module];
		else
			[_testsAppDelegate
			    outputFailure: @"-[observe] with data to read "
					   @"available"
				 inModule: module];

		[_client close];

		[_testsAppDelegate
		    outputTesting: @"-[observe] with closed connection"
			 inModule: module];

		break;
	case 2:
		if (object == _accepted &&
		    [object readIntoBuffer: &buf
				    length: 1] == 0)
			[_testsAppDelegate
			    outputSuccess: @"-[observe] with closed connection"
				 inModule: module];
		else
			[_testsAppDelegate
			    outputFailure: @"-[observe] with closed connection"
				 inModule: module];


		break;
	default:
		OF_ENSURE(0);

	}
}
@end

@implementation TestsAppDelegate (OFKernelEventObserverTests)
- (void)kernelEventObserverTestsWithClass: (Class)class
{
	ObserverTest *test;





	module = [class className];
	test = [[[ObserverTest alloc]



	    initWithTestsAppDelegate: self] autorelease];




	TEST(@"+[observer]",
	    (test->_observer = [OFKernelEventObserver observer]))
	[test->_observer setDelegate: test];

	TEST(@"-[addObjectForReading:]",
	    R([test->_observer addObjectForReading: test->_server]))







	[test run];

















}

- (void)kernelEventObserverTests
{
	OFAutoreleasePool *pool = [[OFAutoreleasePool alloc] init];

#if defined(HAVE_SYS_SELECT_H) || defined(_WIN32)