As-yet untested code for a dispatch source wrapper which can fire multiple event/cancel callbacks

KBMultiDispatchSource.m

#import <Foundation/Foundation.h>
#import <dispatch/dispatch.h>
#import "MAZeroingWeakRef.h"

@interface KBMultiDispatchSource : NSObject
{
	dispatch_source_t		_dispatch;
	dispatch_queue_t		_queue;
	dispatch_source_type_t	_type;
	CFMutableDictionaryRef	_eventObservers;
	CFMutableDictionaryRef	_cancelObservers;
}

- (id) initWithType: (dispatch_source_type_t) type
			 handle: (uintptr_t) handle
			   mask: (unsigned long) mask
		targetQueue: (dispatch_queue_t) targetQueue;

@property (nonatomic, readonly) dispatch_source_type_t type;

- (void) resume;
- (void) suspend;
- (void) cancel;
- (BOOL) testCancel;

- (uintptr_t) handle;
- (unsigned long) mask;

- (void) mergeData: (unsigned long) data;
@property (nonatomic, readonly) unsigned long data;

- (void) resetData;		// sets data to zero-- builds a new dispatch source behind the scenes

- (void) handleEventsForObserver: (id) observer usingBlock: (dispatch_block_t) block;
- (void) handleCancelEventsForObserver: (id) observer usingBlock: (dispatch_block_t) block;
- (void) removeHandlersForObserver: (id) observer;

@end


//
//  KBMultiDispatchSource.m
//  Kobov3
//
//  Created by Jim Dovey on 11-07-07.
//  Copyright 2011 Kobo Inc. All rights reserved.
//

#import "KBMultiDispatchSource.h"

static const void * _CFBlockRetain(CFAllocatorRef allocator, const void * item)
{
	return ( Block_copy(item) );
}

static void _CFBlockRelease(CFAllocatorRef allocator, const void * item)
{
	Block_release(item);
}

@implementation KBMultiDispatchSource

@synthesize type=_type;

+ (void) load
{
	if ( dispatch_barrier_async == 0 )
		return;
	
	// switch in the dispatch_barrier_async() versions since they're available
	
	Method m1 = class_getInstanceMethod(self, @selector(handleEventsForObserver:usingBlock:));
	Method m2 = class_getInstanceMethod(self, @selector(_barrier_handleEventsForObserver:usingBlock:));
	
	if ( m1 != NULL && m2 != NULL )
		method_exchangeImplementations(m1, m2);
	
	m1 = class_getInstanceMethod(self, @selector(handleCancelEventsForObserver:usingBlock:));
	m2 = class_getInstanceMethod(self, @selector(_barrier_handleCancelEventsForObserver:usingBlock:));
	
	if ( m1 != NULL && m2 != NULL )
		method_exchangeImplementations(m1, m2);
	
	m1 = class_getInstanceMethod(self, @selector(removeHandlersForObserver:));
	m2 = class_getInstanceMethod(self, @selector(_barrier_removeHandlersForObserver:));
	
	if ( m1 != NULL && m2 != NULL )
		method_exchangeImplementations(m1, m2);
}

- (void) _setHandlersForSource: (dispatch_source_t) source
{
	{
		MAKE_WEAK_SELF();
		
		dispatch_source_set_event_handler(source, ^{
			USE_WEAK_SELF();
			
			NSDictionary * dict = (NSDictionary *)self->_eventObservers;
			[dict enumerateKeysAndObjectsUsingBlock: ^(id key, id obj, BOOL *stop) {
				dispatch_block_t block = (dispatch_block_t)obj;
				block();
			}];
		});
		
		dispatch_source_set_cancel_handler(source, ^{
			USE_WEAK_SELF();
			
			NSDictionary * dict = (NSDictionary *)self->_cancelObservers;
			[dict enumerateKeysAndObjectsUsingBlock: ^(id key, id obj, BOOL *stop) {
				dispatch_block_t block = (dispatch_block_t)obj;
				block();
			}];
		});
	}
}

- (id) initWithType: (dispatch_source_type_t) type
			 handle: (uintptr_t) handle
			   mask: (unsigned long) mask
		targetQueue: (dispatch_queue_t) targetQueue
{
	self = [super init];
	if ( self == nil )
		return ( nil );
	
	_dispatch = dispatch_source_create(type, handle, mask, targetQueue);
	dispatch_retain(targetQueue);
	_queue = targetQueue;
	_type = type;
	
	CFDictionaryValueCallBacks vcb = {
		.version			= 0,
		.retain				= _CFBlockRetain,
		.release			= _CFBlockRelease,
		.copyDescription	= NULL,
		.equal				= NULL
	};
	
	_eventObservers = CFDictionaryCreateMutable(kCFAllocatorDefault, 0, &kCFTypeDictionaryKeyCallBacks, &vcb);
	_cancelObservers = CFDictionaryCreateMutableCopy(kCFAllocatorDefault, 0, _eventObservers);
	[self _setHandlersForSource: _dispatch];
	
	return ( self );
}

- (void) dealloc
{
	if ( _dispatch != NULL )
		dispatch_release(_dispatch);
	if ( _queue != NULL )
		dispatch_release(_queue);
	if ( _eventObservers != NULL )
		CFRelease(_eventObservers);
	if ( _cancelObservers != NULL )
		CFRelease(_cancelObservers);
	[super dealloc];
}

- (void) resume
{
	dispatch_resume(_dispatch);
}

- (void) suspend
{
	dispatch_suspend(_dispatch);
}

- (void) cancel
{
	dispatch_source_cancel(_dispatch);
}

- (BOOL) testCancel
{
	return ( dispatch_source_testcancel(_dispatch) );
}

- (uintptr_t) handle
{
	return ( dispatch_source_get_handle(_dispatch) );
}

- (unsigned long) mask
{
	return ( dispatch_source_get_mask(_dispatch) );
}

- (void) mergeData: (unsigned long) data
{
	dispatch_source_merge_data(_dispatch, data);
}

- (unsigned long) data
{
	return ( dispatch_source_get_data(_dispatch) );
}

- (void) resetData
{
	dispatch_source_t newSource = dispatch_source_create(_type, dispatch_source_get_handle(_dispatch), dispatch_source_get_mask(_dispatch), _queue);
	[self _setHandlersForSource: newSource];
	
	dispatch_release(_dispatch);
	_dispatch = newSource;
	dispatch_resume(_dispatch);
}

- (void) handleEventsForObserver: (id) observer usingBlock: (dispatch_block_t) block
{
	MAZeroingWeakRef * weakSelfRef = [MAZeroingWeakRef refWithTarget: self];
	dispatch_async(_queue, ^{
		MAZeroingWeakRef * observerRef = [MAZeroingWeakRef refWithTarget: observer];
		[observerRef setCleanupBlock: ^(id target) {
			[[weakSelfRef target] removeHandlersForObserver: target];
		}];
		CFDictionarySetValue(_eventObservers, observer, block);
	});
}

- (void) _barrier_handleEventsForObserver: (id) observer usingBlock: (dispatch_block_t) block
{
	MAZeroingWeakRef * weakSelfRef = [MAZeroingWeakRef refWithTarget: self];
	dispatch_barrier_async(_queue, ^{
		MAZeroingWeakRef * observerRef = [MAZeroingWeakRef refWithTarget: observer];
		[observerRef setCleanupBlock: ^(id target) {
			[[weakSelfRef target] removeHandlersForObserver: target];
		}];
		CFDictionarySetValue(_eventObservers, observer, block);
	});
}

- (void) handleCancelEventsForObserver: (id) observer usingBlock: (dispatch_block_t) block
{
	MAZeroingWeakRef * weakSelfRef = [MAZeroingWeakRef refWithTarget: self];
	dispatch_async(_queue, ^{
		MAZeroingWeakRef * observerRef = [MAZeroingWeakRef refWithTarget: observer];
		[observerRef setCleanupBlock: ^(id target) {
			[[weakSelfRef target] removeHandlersForObserver: target];
		}];
		CFDictionarySetValue(_cancelObservers, observer, block);
	});
}

- (void) _barrier_handleCancelEventsForObserver: (id) observer usingBlock: (dispatch_block_t) block
{
	MAZeroingWeakRef * weakSelfRef = [MAZeroingWeakRef refWithTarget: self];
	dispatch_barrier_async(_queue, ^{
		MAZeroingWeakRef * observerRef = [MAZeroingWeakRef refWithTarget: observer];
		[observerRef setCleanupBlock: ^(id target) {
			[[weakSelfRef target] removeHandlersForObserver: target];
		}];
		CFDictionarySetValue(_cancelObservers, observer, block);
	});
}

- (void) removeHandlersForObserver: (id) observer
{
	dispatch_async(_queue, ^{
		CFDictionaryRemoveValue(_eventObservers, observer);
		CFDictionaryRemoveValue(_cancelObservers, observer);
	});
}

- (void) _barrier_removeHandlersForObserver: (id) observer
{
	dispatch_barrier_async(_queue, ^{
		CFDictionaryRemoveValue(_eventObservers, observer);
		CFDictionaryRemoveValue(_cancelObservers, observer);
	});
}

@end