From 4281baec56947be1c21693ec1309d2fbfc01aa63 Mon Sep 17 00:00:00 2001 From: Robert Sesek Date: Sun, 29 May 2011 16:31:47 -0400 Subject: [PATCH] Move ownership of the read and write stream from NetworkConnection to NetworkCallbackController --- Source/NetworkCallbackController.h | 13 +++++ Source/NetworkCallbackController.mm | 90 ++++++++++++++++++++--------- Source/NetworkConnection.h | 6 -- Source/NetworkConnection.mm | 48 ++++----------- Source/NetworkConnectionPrivate.h | 4 +- 5 files changed, 86 insertions(+), 75 deletions(-) diff --git a/Source/NetworkCallbackController.h b/Source/NetworkCallbackController.h index 6aee8e4..6f149c6 100644 --- a/Source/NetworkCallbackController.h +++ b/Source/NetworkCallbackController.h @@ -36,6 +36,14 @@ class NetworkCallbackController // Closes down the read/write streams. void CloseConnection(); + // Checks whether the write stream is ready for writing. + BOOL WriteStreamCanAcceptBytes(); + + // Writes the string to the write stream. This will block, so be sure to check + // if it can write before calling this. Returns YES if the string was + // successfully written. + BOOL WriteString(NSString* string); + private: // These static methods forward an invocation to the instance methods. The // last void pointer, named |self|, is the instance of this class. @@ -68,6 +76,11 @@ class NetworkCallbackController // The actual socket. CFSocketRef socket_; // Strong. + // The read and write streams that are scheduled on the |runLoop_|. Both are + // weak and are owned by the run loop source. + CFReadStreamRef readStream_; + CFWriteStreamRef writeStream_; + NetworkConnection* connection_; // Weak, owns this. CFRunLoopRef runLoop_; // Weak. }; diff --git a/Source/NetworkCallbackController.mm b/Source/NetworkCallbackController.mm index a9c4cb3..630ff97 100644 --- a/Source/NetworkCallbackController.mm +++ b/Source/NetworkCallbackController.mm @@ -23,7 +23,9 @@ #import "NetworkConnectionPrivate.h" NetworkCallbackController::NetworkCallbackController(NetworkConnection* connection) - : connection_(connection), + : readStream_(NULL), + writeStream_(NULL), + connection_(connection), runLoop_(CFRunLoopGetCurrent()) { } @@ -83,6 +85,43 @@ void NetworkCallbackController::CloseConnection() UnscheduleWriteStream(); } +BOOL NetworkCallbackController::WriteStreamCanAcceptBytes() +{ + return CFWriteStreamCanAcceptBytes(writeStream_); +} + +BOOL NetworkCallbackController::WriteString(NSString* string) +{ + BOOL done = NO; + + char* cString = const_cast([string UTF8String]); + size_t stringLength = strlen(cString); + + // Busy wait while writing. BAADD. Should background this operation. + while (!done) { + if (WriteStreamCanAcceptBytes()) { + // Include the NULL byte in the string when we write. + CFIndex bytesWritten = CFWriteStreamWrite(writeStream_, (UInt8*)cString, stringLength + 1); + if (bytesWritten < 0) { + CFErrorRef error = CFWriteStreamCopyError(writeStream_); + ReportError(error); + break; + } + // Incomplete write. + else if (bytesWritten < static_cast(strlen(cString))) { + // Adjust the buffer and wait for another chance to write. + stringLength -= bytesWritten; + memmove(string, string + bytesWritten, stringLength); + } + else { + done = YES; + } + } + } + + return done; +} + // Static Methods ////////////////////////////////////////////////////////////// void NetworkCallbackController::SocketAcceptCallback(CFSocketRef socket, @@ -116,14 +155,11 @@ void NetworkCallbackController::OnSocketAccept(CFSocketRef socket, CFDataRef address, const void* data) { - CFReadStreamRef readStream; - CFWriteStreamRef writeStream; - // Create the streams on the socket. CFStreamCreatePairWithSocket(kCFAllocatorDefault, *(CFSocketNativeHandle*)data, // Socket handle. - &readStream, // Read stream in-pointer. - &writeStream); // Write stream in-pointer. + &readStream_, // Read stream in-pointer. + &writeStream_); // Write stream in-pointer. // Create struct to register callbacks for the stream. CFStreamClientContext context = { 0 }; @@ -134,15 +170,15 @@ void NetworkCallbackController::OnSocketAccept(CFSocketRef socket, kCFStreamEventHasBytesAvailable | kCFStreamEventErrorOccurred | kCFStreamEventEndEncountered; - if (CFReadStreamSetClient(readStream, readFlags, &NetworkCallbackController::ReadStreamCallback, &context)) + if (CFReadStreamSetClient(readStream_, readFlags, &NetworkCallbackController::ReadStreamCallback, &context)) // Schedule in run loop to do asynchronous communication with the engine. - CFReadStreamScheduleWithRunLoop(readStream, runLoop_, kCFRunLoopCommonModes); + CFReadStreamScheduleWithRunLoop(readStream_, runLoop_, kCFRunLoopCommonModes); else return; // Open the stream now that it's scheduled on the run loop. - if (!CFReadStreamOpen(readStream)) { - ReportError(CFReadStreamCopyError(readStream)); + if (!CFReadStreamOpen(readStream_)) { + ReportError(CFReadStreamCopyError(readStream_)); return; } @@ -151,20 +187,18 @@ void NetworkCallbackController::OnSocketAccept(CFSocketRef socket, kCFStreamEventCanAcceptBytes | kCFStreamEventErrorOccurred | kCFStreamEventEndEncountered; - if (CFWriteStreamSetClient(writeStream, writeFlags, &NetworkCallbackController::WriteStreamCallback, &context)) + if (CFWriteStreamSetClient(writeStream_, writeFlags, &NetworkCallbackController::WriteStreamCallback, &context)) // Schedule it in the run loop to receive error information. - CFWriteStreamScheduleWithRunLoop(writeStream, runLoop_, kCFRunLoopCommonModes); + CFWriteStreamScheduleWithRunLoop(writeStream_, runLoop_, kCFRunLoopCommonModes); else return; // Open the write stream. - if (!CFWriteStreamOpen(writeStream)) { - ReportError(CFWriteStreamCopyError(writeStream)); + if (!CFWriteStreamOpen(writeStream_)) { + ReportError(CFWriteStreamCopyError(writeStream_)); return; } - connection_.readStream = readStream; - connection_.writeStream = writeStream; [connection_ socketDidAccept]; } @@ -174,8 +208,8 @@ void NetworkCallbackController::OnReadStreamEvent(CFReadStreamRef stream, switch (eventType) { case kCFStreamEventHasBytesAvailable: - if (connection_.readStream) - [connection_ readStreamHasData]; + if (readStream_) + [connection_ readStreamHasData:stream]; break; case kCFStreamEventErrorOccurred: @@ -213,22 +247,22 @@ void NetworkCallbackController::OnWriteStreamEvent(CFWriteStreamRef stream, void NetworkCallbackController::UnscheduleReadStream() { - if (!connection_.readStream) + if (!readStream_) return; - CFReadStreamUnscheduleFromRunLoop(connection_.readStream, runLoop_, kCFRunLoopCommonModes); - CFReadStreamClose(connection_.readStream); - CFRelease(connection_.readStream); - connection_.readStream = NULL; + CFReadStreamUnscheduleFromRunLoop(readStream_, runLoop_, kCFRunLoopCommonModes); + CFReadStreamClose(readStream_); + CFRelease(readStream_); + readStream_ = NULL; } void NetworkCallbackController::UnscheduleWriteStream() { - if (!connection_.writeStream) + if (!writeStream_) return; - CFWriteStreamUnscheduleFromRunLoop(connection_.writeStream, runLoop_, kCFRunLoopCommonModes); - CFWriteStreamClose(connection_.writeStream); - CFRelease(connection_.writeStream); - connection_.writeStream = NULL; + CFWriteStreamUnscheduleFromRunLoop(writeStream_, runLoop_, kCFRunLoopCommonModes); + CFWriteStreamClose(writeStream_); + CFRelease(writeStream_); + writeStream_ = NULL; } void NetworkCallbackController::ReportError(CFErrorRef error) diff --git a/Source/NetworkConnection.h b/Source/NetworkConnection.h index 9286714..dd8b621 100644 --- a/Source/NetworkConnection.h +++ b/Source/NetworkConnection.h @@ -46,12 +46,6 @@ class NetworkCallbackController; // Internal class that manages CFNetwork callbacks. Strong. NetworkCallbackController* callbackController_; - // The read stream that is scheduled on the main run loop. Weak. - CFReadStreamRef readStream_; - - // The write stream. Weak. - CFWriteStreamRef writeStream_; - // Run loop source used to quit the thread. Strong. CFRunLoopSourceRef quitSource_; diff --git a/Source/NetworkConnection.mm b/Source/NetworkConnection.mm index 63370a8..c133614 100644 --- a/Source/NetworkConnection.mm +++ b/Source/NetworkConnection.mm @@ -36,11 +36,8 @@ void PerformQuitSignal(void* info) @synthesize port = port_; @synthesize connected = connected_; @synthesize delegate = delegate_; - -@synthesize readStream = readStream_; @synthesize lastReadTransaction = lastReadTransaction_; @synthesize currentPacket = currentPacket_; -@synthesize writeStream = writeStream_; @synthesize lastWrittenTransaction = lastWrittenTransaction_; @synthesize queuedWrites = queuedWrites_; @@ -171,7 +168,7 @@ void PerformQuitSignal(void* info) */ - (void)send:(NSString*)command { - if (lastReadTransaction_ >= lastWrittenTransaction_ && CFWriteStreamCanAcceptBytes(writeStream_)) { + if (lastReadTransaction_ >= lastWrittenTransaction_ && callbackController_->WriteStreamCanAcceptBytes()) { [self performSend:command]; } else { [writeQueueLock_ lock]; @@ -319,12 +316,12 @@ void PerformQuitSignal(void* info) /** * Callback from the CFReadStream that there is data waiting to be read. */ -- (void)readStreamHasData +- (void)readStreamHasData:(CFReadStreamRef)stream { const NSUInteger kBufferSize = 1024; UInt8 buffer[kBufferSize]; CFIndex bufferOffset = 0; // Starting point in |buffer| to work with. - CFIndex bytesRead = CFReadStreamRead(readStream_, buffer, kBufferSize); + CFIndex bytesRead = CFReadStreamRead(stream, buffer, kBufferSize); const char* charBuffer = (const char*)buffer; // The read loop works by going through the buffer until all the bytes have @@ -441,38 +438,13 @@ void PerformQuitSignal(void* info) NSInteger transaction = [self transactionIDFromCommand:command]; if (transaction != NSNotFound && transaction < lastWrittenTransaction_) return; - - BOOL done = NO; - - char* string = (char*)[command UTF8String]; - size_t stringLength = strlen(string); - - // Busy wait while writing. BAADD. Should background this operation. - while (!done) { - if (CFWriteStreamCanAcceptBytes(writeStream_)){ - // Include the NULL byte in the string when we write. - CFIndex bytesWritten = CFWriteStreamWrite(writeStream_, (UInt8*)string, stringLength + 1); - if (bytesWritten < 0) { - CFErrorRef error = CFWriteStreamCopyError(writeStream_); - NSLog(@"Write stream error: %@", error); - CFRelease(error); - } - // Incomplete write. - else if (bytesWritten < static_cast(strlen(string))) { - // Adjust the buffer and wait for another chance to write. - stringLength -= bytesWritten; - memmove(string, string + bytesWritten, stringLength); - } - else { - done = YES; - // We need to scan the string to find the transactionID. - if (transaction == NSNotFound) { - NSLog(@"sent %@ without a transaction ID", command); - continue; - } - lastWrittenTransaction_ = transaction; - } + + if (callbackController_->WriteString(command)) { + // We need to scan the string to find the transactionID. + if (transaction == NSNotFound) { + NSLog(@"sent %@ without a transaction ID", command); } + lastWrittenTransaction_ = transaction; } // Log this trancation. @@ -495,7 +467,7 @@ void PerformQuitSignal(void* info) // We don't want to block because this is called from the main thread. // |-performSend:| busy waits when the stream is not ready. Bail out // before we do that becuase busy waiting is BAD. - if (CFWriteStreamCanAcceptBytes(writeStream_)) { + if (callbackController_->WriteStreamCanAcceptBytes()) { [self performSend:command]; [queuedWrites_ removeObjectAtIndex:0]; } diff --git a/Source/NetworkConnectionPrivate.h b/Source/NetworkConnectionPrivate.h index 0cffb27..02090ad 100644 --- a/Source/NetworkConnectionPrivate.h +++ b/Source/NetworkConnectionPrivate.h @@ -20,10 +20,8 @@ // by the C++ NetworkCallbackController to communicate. @interface NetworkConnection () -@property (assign) CFReadStreamRef readStream; @property NSInteger lastReadTransaction; @property (retain) NSMutableString* currentPacket; -@property (assign) CFWriteStreamRef writeStream; @property NSInteger lastWrittenTransaction; @property (retain) NSMutableArray* queuedWrites; @@ -31,7 +29,7 @@ - (void)socketDidAccept; - (void)socketDisconnected; -- (void)readStreamHasData; +- (void)readStreamHasData:(CFReadStreamRef)stream; // These methods MUST be called on the network thread as they are not threadsafe. - (void)send:(NSString*)command; -- 2.22.5