Adding receiveOnce operation to GCDAsyncUdpSocket. So there are now two modes of operation: one-at-a-time & continuous receive.

robbiehanson · robbiehanson · commit 980bbd685b40 · 2012-04-08T18:00:02.000-07:00
diff --git a/GCD/GCDAsyncUdpSocket.h b/GCD/GCDAsyncUdpSocket.h
@@ -469,14 +469,54 @@ typedef BOOL (^GCDAsyncUdpSocketReceiveFilterBlock)(NSData *data, NSData *addres
 #pragma mark Receiving
 
 /**
- * Begins receiving udp packets on the socket.
- * It will continue to receive packets until the socket is closed, or until pauseReceiving is called.
+ * There are two modes of operation for receiving packets: one-at-a-time & continuous.
+ * 
+ * In one-at-a-time mode, you call receiveOnce everytime your delegate is ready to process an incoming udp packet.
+ * Receiving packets one-at-a-time may be better suited for implementing certain state machine code,
+ * where your state machine may not always be ready to process incoming packets.
+ * 
+ * In continuous mode, the delegate is invoked immediately everytime incoming udp packets are received.
+ * Receiving packets continuously is better suited to real-time streaming applications.
+ * 
+ * You may switch back and forth between one-at-a-time mode and continuous mode.
+ * If the socket is currently in continuous mode, calling this method will switch it to one-at-a-time mode.
+ * 
+ * When a packet is received (and not filtered by the optional receive filter),
+ * the delegate method (udpSocket:didReceiveData:fromAddress:withFilterContext:) is invoked.
+ * 
+ * If the socket is able to begin receiving packets, this method returns YES.
+ * Otherwise it returns NO, and sets the errPtr with appropriate error information.
+ * 
+ * An example error:
+ * You created a udp socket to act as a server, and immediately called receive.
+ * You forgot to first bind the socket to a port number, and received a error with a message like:
+ * "Must bind socket before you can receive data."
+**/
+- (BOOL)receiveOnce:(NSError **)errPtr;
+
+/**
+ * There are two modes of operation for receiving packets: one-at-a-time & continuous.
+ * 
+ * In one-at-a-time mode, you call receiveOnce everytime your delegate is ready to process an incoming udp packet.
+ * Receiving packets one-at-a-time may be better suited for implementing certain state machine code,
+ * where your state machine may not always be ready to process incoming packets.
+ * 
+ * In continuous mode, the delegate is invoked immediately everytime incoming udp packets are received.
+ * Receiving packets continuously is better suited to real-time streaming applications.
+ * 
+ * You may switch back and forth between one-at-a-time mode and continuous mode.
+ * If the socket is currently in one-at-a-time mode, calling this method will switch it to continuous mode.
  * 
  * For every received packet (not filtered by the optional receive filter),
  * the delegate method (udpSocket:didReceiveData:fromAddress:withFilterContext:) is invoked.
  * 
  * If the socket is able to begin receiving packets, this method returns YES.
  * Otherwise it returns NO, and sets the errPtr with appropriate error information.
+ * 
+ * An example error:
+ * You created a udp socket to act as a server, and immediately called receive.
+ * You forgot to first bind the socket to a port number, and received a error with a message like:
+ * "Must bind socket before you can receive data."
 **/
 - (BOOL)beginReceiving:(NSError **)errPtr;
 
diff --git a/GCD/GCDAsyncUdpSocket.m b/GCD/GCDAsyncUdpSocket.m
@@ -110,20 +110,21 @@
 	kDidBind                 = 1 <<  1,  // If set, bind has been called.
 	kConnecting              = 1 <<  2,  // If set, a connection attempt is in progress.
 	kDidConnect              = 1 <<  3,  // If set, socket is connected.
-	kReceive                 = 1 <<  4,  // If set, receive is enabled
-	kIPv4Deactivated         = 1 <<  5,  // If set, socket4 was closed due to bind or connect on IPv6.
-	kIPv6Deactivated         = 1 <<  6,  // If set, socket6 was closed due to bind or connect on IPv4.
-	kSend4SourceSuspended    = 1 <<  7,  // If set, send4Source is suspended.
-	kSend6SourceSuspended    = 1 <<  8,  // If set, send6Source is suspended.
-	kReceive4SourceSuspended = 1 <<  9,  // If set, receive4Source is suspended.
-	kReceive6SourceSuspended = 1 << 10,  // If set, receive6Source is suspended.
-	kSock4CanAcceptBytes     = 1 << 11,  // If set, we know socket4 can accept bytes. If unset, it's unknown.
-	kSock6CanAcceptBytes     = 1 << 12,  // If set, we know socket6 can accept bytes. If unset, it's unknown.
-	kForbidSendReceive       = 1 << 13,  // If set, no new send or receive operations are allowed to be queued.
-	kCloseAfterSends         = 1 << 14,  // If set, close as soon as no more sends are queued.
-	kFlipFlop                = 1 << 15,  // Used to alternate between IPv4 and IPv6 sockets.
+	kReceiveOnce             = 1 <<  4,  // If set, one-at-a-time receive is enabled
+	kReceiveContinuous       = 1 <<  5,  // If set, continuous receive is enabled
+	kIPv4Deactivated         = 1 <<  6,  // If set, socket4 was closed due to bind or connect on IPv6.
+	kIPv6Deactivated         = 1 <<  7,  // If set, socket6 was closed due to bind or connect on IPv4.
+	kSend4SourceSuspended    = 1 <<  8,  // If set, send4Source is suspended.
+	kSend6SourceSuspended    = 1 <<  9,  // If set, send6Source is suspended.
+	kReceive4SourceSuspended = 1 << 10,  // If set, receive4Source is suspended.
+	kReceive6SourceSuspended = 1 << 11,  // If set, receive6Source is suspended.
+	kSock4CanAcceptBytes     = 1 << 12,  // If set, we know socket4 can accept bytes. If unset, it's unknown.
+	kSock6CanAcceptBytes     = 1 << 13,  // If set, we know socket6 can accept bytes. If unset, it's unknown.
+	kForbidSendReceive       = 1 << 14,  // If set, no new send or receive operations are allowed to be queued.
+	kCloseAfterSends         = 1 << 15,  // If set, close as soon as no more sends are queued.
+	kFlipFlop                = 1 << 16,  // Used to alternate between IPv4 and IPv6 sockets.
 #if TARGET_OS_IPHONE
-	kAddedStreamListener     = 1 << 16,  // If set, CFStreams have been added to listener thread
+	kAddedStreamListener     = 1 << 17,  // If set, CFStreams have been added to listener thread
 #endif
 };
 
@@ -170,6 +171,8 @@ @interface GCDAsyncUdpSocket ()
 	unsigned long socket4FDBytesAvailable;
 	unsigned long socket6FDBytesAvailable;
 	
+	uint32_t pendingFilterOperations;
+	
 	NSData   *cachedLocalAddress4;
 	NSString *cachedLocalHost4;
 	uint16_t  cachedLocalPort4;
@@ -3805,6 +3808,52 @@ - (void)setupSendTimerWithTimeout:(NSTimeInterval)timeout
 #pragma mark Receiving
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
+- (BOOL)receiveOnce:(NSError **)errPtr
+{
+	LogTrace();
+	
+	__block BOOL result = NO;
+	__block NSError *err = nil;
+	
+	dispatch_block_t block = ^{
+		
+		if ((flags & kReceiveOnce) == 0)
+		{
+			if ((flags & kDidCreateSockets) == 0)
+			{
+				NSString *msg = @"Must bind socket before you can receive data. "
+				@"You can do this explicitly via bind, or implicitly via connect or by sending data.";
+				
+				err = [self badConfigError:msg];
+				return_from_block;
+			}
+			
+			flags |=  kReceiveOnce;       // Enable
+			flags &= ~kReceiveContinuous; // Disable
+			
+			dispatch_async(socketQueue, ^{ @autoreleasepool {
+				
+				[self doReceive];
+			}});
+		}
+		
+		result = YES;
+	};
+	
+	if (dispatch_get_current_queue() == socketQueue)
+		block();
+	else
+		dispatch_sync(socketQueue, block);
+	
+	if (err)
+		LogError(@"Error in beginReceiving: %@", err);
+	
+	if (errPtr)
+		*errPtr = err;
+	
+	return result;
+}
+
 - (BOOL)beginReceiving:(NSError **)errPtr
 {
 	LogTrace();
@@ -3814,7 +3863,7 @@ - (BOOL)beginReceiving:(NSError **)errPtr
 	
 	dispatch_block_t block = ^{
 		
-		if ((flags & kReceive) == 0)
+		if ((flags & kReceiveContinuous) == 0)
 		{
 			if ((flags & kDidCreateSockets) == 0)
 			{
@@ -3825,7 +3874,8 @@ - (BOOL)beginReceiving:(NSError **)errPtr
 				return_from_block;
 			}
 			
-			flags |= kReceive;
+			flags |= kReceiveContinuous; // Enable
+			flags &= ~kReceiveOnce;      // Disable
 			
 			dispatch_async(socketQueue, ^{ @autoreleasepool {
 				
@@ -3856,7 +3906,8 @@ - (void)pauseReceiving
 	
 	dispatch_block_t block = ^{
 		
-		flags &= ~kReceive;
+		flags &= ~kReceiveOnce;       // Disable
+		flags &= ~kReceiveContinuous; // Disable
 		
 		if (socket4FDBytesAvailable > 0) {
 			[self suspendReceive4Source];
@@ -3911,7 +3962,7 @@ - (void)doReceive
 {
 	LogTrace();
 	
-	if ((flags & kReceive) == 0)
+	if ((flags & (kReceiveOnce | kReceiveContinuous)) == 0)
 	{
 		LogVerbose(@"Receiving is paused...");
 		
@@ -3925,6 +3976,20 @@ - (void)doReceive
 		return;
 	}
 	
+	if ((flags & kReceiveOnce) && (pendingFilterOperations > 0))
+	{
+		LogVerbose(@"Receiving is temporarily paused (pending filter operations)...");
+		
+		if (socket4FDBytesAvailable > 0) {
+			[self suspendReceive4Source];
+		}
+		if (socket6FDBytesAvailable > 0) {
+			[self suspendReceive6Source];
+		}
+		
+		return;
+	}
+	
 	if ((socket4FDBytesAvailable == 0) && (socket6FDBytesAvailable == 0))
 	{
 		LogVerbose(@"No data available to receive...");
@@ -4053,6 +4118,8 @@ - (void)doReceive
 	
 	
 	BOOL waitingForSocket = NO;
+	BOOL ignoredDueToAddress = NO;
+	
 	NSError *error = nil;
 	
 	if (result == 0)
@@ -4068,34 +4135,56 @@ - (void)doReceive
 	}
 	else
 	{
-		BOOL ignored = NO;
-		
 		if (flags & kDidConnect)
 		{
 			if (addr4 && ![self isConnectedToAddress4:addr4])
-				ignored = YES;
+				ignoredDueToAddress = YES;
 			if (addr6 && ![self isConnectedToAddress6:addr6])
-				ignored = YES;
+				ignoredDueToAddress = YES;
 		}
 		
 		NSData *addr = (addr4 != nil) ? addr4 : addr6;
 		
-		if (!ignored)
+		if (!ignoredDueToAddress)
 		{
 			if (filterBlock && filterQueue)
 			{
 				// Run data through filter, and if approved, notify delegate
+				pendingFilterOperations++;
+				
 				dispatch_async(filterQueue, ^{ @autoreleasepool {
 					
 					id filterContext = nil;
+					BOOL allowed = filterBlock(data, addr, &filterContext);
 					
-					if (filterBlock(data, addr, &filterContext))
-					{
-						// Transition back to socketQueue to get the current delegate / delegateQueue
-						dispatch_async(socketQueue, ^{
+					// Transition back to socketQueue to get the current delegate / delegateQueue
+					
+					dispatch_async(socketQueue, ^{ @autoreleasepool {
+						
+						pendingFilterOperations--;
+						
+						if (allowed)
+						{
 							[self notifyDidReceiveData:data fromAddress:addr withFilterContext:filterContext];
-						});
-					}
+						}
+						
+						if (flags & kReceiveOnce)
+						{
+							if (allowed)
+							{
+								// The delegate has been notified,
+								// so our receive once operation has completed.
+								flags &= ~kReceiveOnce;
+							}
+							else if (pendingFilterOperations == 0)
+							{
+								// All pending filter operations have completed,
+								// and none were allowed through.
+								// Our receive once operation hasn't completed yet.
+								[self doReceive];
+							}
+						}
+					}});
 					
 				}});
 			}
@@ -4124,7 +4213,25 @@ - (void)doReceive
 	}
 	else
 	{
-		[self doReceive];
+		if (flags & kReceiveContinuous)
+		{
+			// Continuous receive mode
+			[self doReceive];
+		}
+		else
+		{
+			// One-at-a-time receive mode
+			if (ignoredDueToAddress)
+			{
+				[self doReceive];
+			}
+			else if (pendingFilterOperations == 0)
+			{
+				// The delegate has been notified (no set filter).
+				// So our receive once operation has completed.
+				flags &= ~kReceiveOnce;
+			}
+		}
 	}
 }
 
