ObjFW  Diff

	OF_ENSURE(write(cancelFD[1], "", 1) > 0);
#else
	OF_ENSURE(sendto(cancelFD[1], "", 1, 0, (struct sockaddr*)&cancelAddr,
	    sizeof(cancelAddr)) > 0);
#endif
}

- (void)_addFileDescriptorForReading: (int)fd
- (void)OF_addFileDescriptorForReading: (int)fd
{
	@throw [OFNotImplementedException exceptionWithClass: [self class]
						    selector: _cmd];
}

- (void)OF_addFileDescriptorForWriting: (int)fd
{
	@throw [OFNotImplementedException exceptionWithClass: [self class]
						    selector: _cmd];
}

- (void)_addFileDescriptorForWriting: (int)fd
- (void)OF_removeFileDescriptorForReading: (int)fd
{
	@throw [OFNotImplementedException exceptionWithClass: [self class]
						    selector: _cmd];
}

- (void)_removeFileDescriptorForReading: (int)fd
- (void)OF_removeFileDescriptorForWriting: (int)fd
{
	@throw [OFNotImplementedException exceptionWithClass: [self class]
						    selector: _cmd];
}

- (void)_removeFileDescriptorForWriting: (int)fd
{
	@throw [OFNotImplementedException exceptionWithClass: [self class]
						    selector: _cmd];
}

- (void)_processQueue
- (void)OF_processQueue
{
	[mutex lock];
	@try {
		OFStream **queueObjects = [queue objects];
		int *queueInfoCArray = [queueInfo cArray];
		int *queueFDsCArray = [queueFDs cArray];
		size_t i, count = [queue count];

				FDToStream[fd] = nil;
			}

			switch (action) {
			case QUEUE_ADD | QUEUE_READ:
				[readStreams addObject: stream];

				[self _addFileDescriptorForReading: fd];
				[self OF_addFileDescriptorForReading: fd];

				break;
			case QUEUE_ADD | QUEUE_WRITE:
				[writeStreams addObject: stream];

				[self _addFileDescriptorForWriting: fd];
				[self OF_addFileDescriptorForWriting: fd];

				break;
			case QUEUE_REMOVE | QUEUE_READ:
				[readStreams removeObjectIdenticalTo: stream];

				[self _removeFileDescriptorForReading: fd];
				[self OF_removeFileDescriptorForReading: fd];

				break;
			case QUEUE_REMOVE | QUEUE_WRITE:
				[writeStreams removeObjectIdenticalTo: stream];

				[self _removeFileDescriptorForWriting: fd];
				[self OF_removeFileDescriptorForWriting: fd];

				break;
			default:
				assert(0);
			}
		}

	OF_ENSURE(write(cancelFD[1], "", 1) > 0);
#else
	OF_ENSURE(sendto(cancelFD[1], "", 1, 0, (struct sockaddr*)&cancelAddr,
	    sizeof(cancelAddr)) > 0);
#endif
}

- (BOOL)_processCache
- (BOOL)OF_processCache
{
	OFStream **objects = [readStreams objects];
	size_t i, count = [readStreams count];
	BOOL foundInCache = NO;


	for (i = 0; i < count; i++) {

		if ([objects[i] pendingBytes] > 0 &&
		    ![objects[i] _isWaitingForDelimiter]) {
		    ![objects[i] OF_isWaitingForDelimiter]) {
			void *pool = objc_autoreleasePoolPush();
			[delegate streamIsReadyForReading: objects[i]];
			foundInCache = YES;
			objc_autoreleasePoolPop(pool);
		}
	}