ObjFW  Diff

}

- (void)setDelegate: (id <OFStreamObserverDelegate>)delegate_
{
	OF_SETTER(delegate, delegate_, YES, NO)
}

- (void)addStreamToObserveForReading: (OFStream*)stream
- (void)addStreamForReading: (OFStream*)stream
{
	OFAutoreleasePool *pool = [[OFAutoreleasePool alloc] init];
	OFNumber *qi = [OFNumber numberWithInt: QUEUE_ADD | QUEUE_READ];

	@synchronized (queue) {
		[queue addObject: stream];
		[queueInfo addObject: qi];
	}

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

	[pool release];
}

- (void)addStreamToObserveForWriting: (OFStream*)stream
- (void)addStreamForWriting: (OFStream*)stream
{
	OFAutoreleasePool *pool = [[OFAutoreleasePool alloc] init];
	OFNumber *qi = [OFNumber numberWithInt: QUEUE_ADD | QUEUE_WRITE];

	@synchronized (queue) {
		[queue addObject: stream];
		[queueInfo addObject: qi];
	}

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

	[pool release];
}

- (void)removeStreamToObserveForReading: (OFStream*)stream
- (void)removeStreamForReading: (OFStream*)stream
{
	OFAutoreleasePool *pool = [[OFAutoreleasePool alloc] init];
	OFNumber *qi = [OFNumber numberWithInt: QUEUE_REMOVE | QUEUE_READ];

	@synchronized (queue) {
		[queue addObject: stream];
		[queueInfo addObject: qi];
	}

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

	[pool release];
}

- (void)removeStreamToObserveForWriting: (OFStream*)stream
- (void)removeStreamForWriting: (OFStream*)stream
{
	OFAutoreleasePool *pool = [[OFAutoreleasePool alloc] init];
	OFNumber *qi = [OFNumber numberWithInt: QUEUE_REMOVE | QUEUE_WRITE];

	@synchronized (queue) {
		[queue addObject: stream];
		[queueInfo addObject: qi];
	}

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

	[pool release];
}

- (void)_addStreamToObserveForReading: (OFStream*)strea;
- (void)_addStreamForReading: (OFStream*)strea;
{
	@throw [OFNotImplementedException newWithClass: isa
					      selector: _cmd];
}

- (void)_addStreamForWriting: (OFStream*)stream
{
	@throw [OFNotImplementedException newWithClass: isa
					      selector: _cmd];
}

- (void)_addStreamToObserveForWriting: (OFStream*)stream
- (void)_removeStreamForReading: (OFStream*)stream
{
	@throw [OFNotImplementedException newWithClass: isa
					      selector: _cmd];
}

- (void)_removeStreamToObserveForReading: (OFStream*)stream
{
	@throw [OFNotImplementedException newWithClass: isa
					      selector: _cmd];
}

- (void)_removeStreamToObserveForWriting: (OFStream*)stream
- (void)_removeStreamForWriting: (OFStream*)stream
{
	@throw [OFNotImplementedException newWithClass: isa
					      selector: _cmd];
}

- (void)_processQueue
{

				FDToStream[fd] = nil;
			}

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

				[self _addStreamToObserveForReading:
				    queueCArray[i]];
				[self _addStreamForReading: queueCArray[i]];

				break;
			case QUEUE_ADD | QUEUE_WRITE:
				[writeStreams addObject: queueCArray[i]];

				[self _addStreamToObserveForWriting:
				    queueCArray[i]];
				[self _addStreamForWriting: queueCArray[i]];

				break;
			case QUEUE_REMOVE | QUEUE_READ:
				[readStreams removeObjectIdenticalTo:
				    queueCArray[i]];

				[self _removeStreamToObserveForReading:
				    queueCArray[i]];
				[self _removeStreamForReading: queueCArray[i]];

				break;
			case QUEUE_REMOVE | QUEUE_WRITE:
				[writeStreams removeObjectIdenticalTo:
				    queueCArray[i]];

				[self _removeStreamToObserveForWriting:
				    queueCArray[i]];
				[self _removeStreamForWriting: queueCArray[i]];

				break;
			default:
				assert(0);
			}
		}

	size_t i, count = [readStreams count];
	BOOL foundInCache = NO;

	pool = [[OFAutoreleasePool alloc] init];

	for (i = 0; i < count; i++) {
		if ([cArray[i] pendingBytes] > 0) {
			[delegate streamDidBecomeReadyForReading: cArray[i]];
			[delegate streamIsReadyForReading: cArray[i]];
			foundInCache = YES;
			[pool releaseObjects];
		}
	}

	[pool release];

	/*
	 * As long as we have data in the cache for any stream, we don't want
	 * to block.
	 */
	if (foundInCache)
		return YES;

	return NO;
}
@end

@implementation OFObject (OFStreamObserverDelegate)
- (void)streamDidBecomeReadyForReading: (OFStream*)stream
- (void)streamIsReadyForReading: (OFStream*)stream
{
}

- (void)streamDidBecomeReadyForWriting: (OFStream*)stream
- (void)streamIsReadyForWriting: (OFStream*)stream
{
}

- (void)streamDidReceiveException: (OFStream*)stream
{
}
@end