Source/MessageQueue.m

   1 /*
   2  * MacGDBp
   3  * Copyright (c) 2013, Blue Static <http://www.bluestatic.org>
   4  *
   5  * This program is free software; you can redistribute it and/or modify it under the terms of the GNU
   6  * General Public License as published by the Free Software Foundation; either version 2 of the
   7  * License, or (at your option) any later version.
   8  *
   9  * This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without
  10  * even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  11  * General Public License for more details.
  12  *
  13  * You should have received a copy of the GNU General Public License along with this program; if not,
  14  * write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
  15  */
  16
  17 #import "MessageQueue.h"
  18
  19 #include <netinet/in.h>
  20 #include <sys/socket.h>
  21
  22 @interface MessageQueue (Private)
  23 // Thread main function that is started from -connect.
  24 - (void)runMessageQueue;
  25
  26 // All the following methods must be called from the -runMessageQueue thread.
  27
  28 // Creates a listening socket and schedules it in the run loop.
  29 - (void)listenForClient;
  30
  31 // Closes down the listening socket, the child socket, and the streams.
  32 - (void)disconnectClient;
  33
  34 // This first calls -disconnectClient and then stops the run loop and terminates
  35 // the -runMessageQueue thread.
  36 - (void)stopRunLoop;
  37
  38 // Adds a |message| to |_queue|.
  39 - (void)enqueueMessage:(NSString*)message;
  40
  41 // If the write stream is ready and there is data to send, sends the next message.
  42 - (void)dequeueAndSend;
  43
  44 // Writes the string into the write stream.
  45 - (void)performSend:(NSString*)message;
  46
  47 // Reads bytes out of the read stream. This may be called multiple times if the
  48 // message cannot be read in one pass.
  49 - (void)readMessageFromStream;
  50
  51 // Converts a CFErrorRef to an NSError and passes it to the delegate.
  52 - (void)reportError:(CFErrorRef)error;
  53
  54 // Forwarding methods from the CoreFoundation callbacks.
  55 - (void)listenSocket:(CFSocketRef)socket acceptedSocket:(CFSocketNativeHandle)child;
  56 - (void)readStream:(CFReadStreamRef)stream handleEvent:(CFStreamEventType)event;
  57 - (void)writeStream:(CFWriteStreamRef)stream handleEvent:(CFStreamEventType)event;
  58 @end
  59
  60 // CoreFoundation Callbacks ////////////////////////////////////////////////////
  61
  62 static void MessageQueueSocketAccept(CFSocketRef socket,
  63                                      CFSocketCallBackType callbackType,
  64                                      CFDataRef address,
  65                                      const void* data,
  66                                      void* self)
  67 {
  68   CFSocketNativeHandle child = *(CFSocketNativeHandle*)data;
  69   [(MessageQueue*)self listenSocket:socket acceptedSocket:child];
  70 }
  71
  72 static void MessageQueueReadEvent(CFReadStreamRef stream,
  73                                   CFStreamEventType eventType,
  74                                   void* self)
  75 {
  76   [(MessageQueue*)self readStream:stream handleEvent:eventType];
  77 }
  78
  79 static void MessageQueueWriteEvent(CFWriteStreamRef stream,
  80                                    CFStreamEventType eventType,
  81                                    void* self)
  82 {
  83   [(MessageQueue*)self writeStream:stream handleEvent:eventType];
  84 }
  85
  86 ////////////////////////////////////////////////////////////////////////////////
  87
  88 @implementation MessageQueue
  89
  90 - (id)initWithPort:(NSUInteger)port delegate:(id<MessageQueueDelegate>)delegate {
  91   if ((self = [super init])) {
  92     _port = port;
  93     _queue = [[NSMutableArray alloc] init];
  94     _delegate = (BSProtocolThreadInvoker<MessageQueueDelegate>*)
  95         [[BSProtocolThreadInvoker alloc] initWithObject:delegate
  96                                          protocol:@protocol(MessageQueueDelegate)
  97                                            thread:[NSThread currentThread]];
  98   }
  99   return self;
 100 }
 101
 102 - (void)dealloc {
 103   [_queue release];
 104   [_delegate release];
 105   [super dealloc];
 106 }
 107
 108 - (BOOL)isConnected {
 109   return _connected;
 110 }
 111
 112 - (void)connect {
 113   if (_thread)
 114     return;
 115
 116   [NSThread detachNewThreadSelector:@selector(runMessageQueue)
 117                            toTarget:self
 118                          withObject:nil];
 119 }
 120
 121 - (void)disconnect {
 122   [self performSelector:@selector(stopRunLoop)
 123                onThread:_thread
 124              withObject:nil
 125           waitUntilDone:NO];
 126 }
 127
 128 - (void)sendMessage:(NSString*)message {
 129   [self performSelector:@selector(enqueueMessage:)
 130                onThread:_thread
 131              withObject:message
 132           waitUntilDone:NO];
 133 }
 134
 135 // Private /////////////////////////////////////////////////////////////////////
 136
 137 - (void)runMessageQueue {
 138   @autoreleasepool {
 139     _thread = [NSThread currentThread];
 140     _runLoop = [NSRunLoop currentRunLoop];
 141
 142     _connected = NO;
 143     _shouldQuit = NO;
 144     [self scheduleListenSocket];
 145
 146     // Use CFRunLoop instead of NSRunLoop because the latter has no programmatic
 147     // stop routine.
 148     CFRunLoopRun();
 149
 150     _thread = nil;
 151     _runLoop = nil;
 152   }
 153 }
 154
 155 - (void)scheduleListenSocket {
 156   // Create the address structure.
 157   struct sockaddr_in address;
 158   memset(&address, 0, sizeof(address));
 159   address.sin_len = sizeof(address);
 160   address.sin_family = AF_INET;
 161   address.sin_port = htons(_port);
 162   address.sin_addr.s_addr = htonl(INADDR_ANY);
 163
 164   // Create the socket signature.
 165   CFSocketSignature signature;
 166   signature.protocolFamily = PF_INET;
 167   signature.socketType = SOCK_STREAM;
 168   signature.protocol = IPPROTO_TCP;
 169   signature.address = (CFDataRef)[NSData dataWithBytes:&address length:sizeof(address)];
 170
 171   CFSocketContext context = { 0 };
 172   context.info = self;
 173
 174   do {
 175     _socket =
 176         CFSocketCreateWithSocketSignature(kCFAllocatorDefault,
 177                                           &signature,  // Socket signature.
 178                                           kCFSocketAcceptCallBack,  // Callback types.
 179                                           &MessageQueueSocketAccept,  // Callback function.
 180                                           &context);  // Context to pass to callout.
 181     if (!_socket) {
 182       // Pump the run loop while waiting for the socket to be reusued. If told
 183       // to quit while waiting, then break out of the loop.
 184       if (CFRunLoopRunInMode(kCFRunLoopDefaultMode, 1, FALSE) && _shouldQuit)
 185         return;
 186       NSLog(@"Could not open socket");
 187       //[connection_ errorEncountered:@"Could not open socket."];
 188     }
 189   } while (!_socket);
 190
 191   // Allow old, yet-to-be recycled sockets to be reused.
 192   int yes = 1;
 193   setsockopt(CFSocketGetNative(_socket), SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int));
 194   setsockopt(CFSocketGetNative(_socket), SOL_SOCKET, SO_REUSEPORT, &yes, sizeof(int));
 195
 196   // Schedule the socket on the run loop.
 197   CFRunLoopSourceRef source = CFSocketCreateRunLoopSource(kCFAllocatorDefault, _socket, 0);
 198   CFRunLoopAddSource([_runLoop getCFRunLoop], source, kCFRunLoopCommonModes);
 199   CFRelease(source);
 200 }
 201
 202 - (void)disconnectClient {
 203   if (_readStream) {
 204     CFReadStreamUnscheduleFromRunLoop(_readStream, [_runLoop getCFRunLoop], kCFRunLoopCommonModes);
 205     CFReadStreamClose(_readStream);
 206     CFRelease(_readStream);
 207     _readStream = NULL;
 208   }
 209
 210   if (_writeStream) {
 211     CFWriteStreamUnscheduleFromRunLoop(_writeStream, [_runLoop getCFRunLoop], kCFRunLoopCommonModes);
 212     CFWriteStreamClose(_writeStream);
 213     CFRelease(_writeStream);
 214     _writeStream = NULL;
 215   }
 216
 217   if (_child) {
 218     close(_child);
 219     _child = NULL;
 220   }
 221
 222   _connected = NO;
 223   [_delegate messageQueueDidDisconnect:self];
 224 }
 225
 226 - (void)stopRunLoop {
 227   _shouldQuit = YES;
 228   [self disconnectClient];
 229   CFRunLoopStop([_runLoop getCFRunLoop]);
 230 }
 231
 232 - (void)enqueueMessage:(NSString*)message {
 233   [_queue addObject:message];
 234   [self dequeueAndSend];
 235 }
 236
 237 - (void)dequeueAndSend {
 238   if (![_queue count])
 239     return;
 240
 241   if (!CFWriteStreamCanAcceptBytes(_writeStream))
 242     return;
 243
 244   NSString* message = [_queue objectAtIndex:0];
 245   [self performSend:message];
 246   [_queue removeObjectAtIndex:0];
 247 }
 248
 249 - (void)performSend:(NSString*)message {
 250   // TODO: May need to negotiate with the server as to the string encoding.
 251   const NSStringEncoding kEncoding = NSUTF8StringEncoding;
 252   // Add space for the NUL byte.
 253   NSUInteger maxBufferSize = [message maximumLengthOfBytesUsingEncoding:kEncoding] + 1;
 254
 255   UInt8* buffer = malloc(maxBufferSize);
 256   bzero(buffer, maxBufferSize);
 257
 258   NSUInteger bufferSize = 0;
 259   if (![message getBytes:buffer
 260                maxLength:maxBufferSize
 261               usedLength:&bufferSize
 262                 encoding:kEncoding
 263                  options:0
 264                    range:NSMakeRange(0, [message length])
 265           remainingRange:NULL]) {
 266     free(buffer);
 267     return;
 268   }
 269
 270   // Include a NUL byte.
 271   ++bufferSize;
 272
 273   // Write the packet out, and spin in a busy wait loop if the stream is not ready. This
 274   // method is only ever called in response to a stream ready event.
 275   NSUInteger totalWritten = 0;
 276   while (totalWritten < bufferSize) {
 277     CFIndex bytesWritten = CFWriteStreamWrite(_writeStream, buffer + totalWritten, bufferSize - totalWritten);
 278     if (bytesWritten < 0) {
 279       [self reportError:CFWriteStreamCopyError(_writeStream)];
 280       break;
 281     }
 282     totalWritten += bytesWritten;
 283   }
 284
 285   [_delegate messageQueue:self didSendMessage:message];
 286
 287   free(buffer);
 288 }
 289
 290 - (void)readMessageFromStream {
 291   const NSUInteger kBufferSize = 1024;
 292   UInt8 buffer[kBufferSize];
 293   CFIndex bufferOffset = 0;  // Starting point in |buffer| to work with.
 294   CFIndex bytesRead = CFReadStreamRead(_readStream, buffer, kBufferSize);
 295   const char* charBuffer = (const char*)buffer;
 296
 297   // The read loop works by going through the buffer until all the bytes have
 298   // been processed.
 299   while (bufferOffset < bytesRead) {
 300     // Find the NUL separator, or the end of the string.
 301     NSUInteger partLength = 0;
 302     for (CFIndex i = bufferOffset; i < bytesRead && charBuffer[i] != '\0'; ++i, ++partLength) ;
 303
 304     // If there is not a current packet, set some state.
 305     if (!_message) {
 306       // Read the message header: the size.  This will be |partLength| bytes.
 307       _totalMessageSize = atoi(charBuffer + bufferOffset);
 308       _messageSize = 0;
 309       _message = [[NSMutableString alloc] initWithCapacity:_totalMessageSize];
 310       bufferOffset += partLength + 1;  // Pass over the NUL byte.
 311       continue;  // Spin the loop to begin reading actual data.
 312     }
 313
 314     // Substring the byte stream and append it to the packet string.
 315     CFStringRef bufferString = CFStringCreateWithBytes(kCFAllocatorDefault,
 316                                                        buffer + bufferOffset,  // Byte pointer, offset by start index.
 317                                                        partLength,  // Length.
 318                                                        kCFStringEncodingUTF8,
 319                                                        true);
 320     [_message appendString:(NSString*)bufferString];
 321     CFRelease(bufferString);
 322
 323     // Advance counters.
 324     _messageSize += partLength;
 325     bufferOffset += partLength + 1;
 326
 327     // If this read finished the packet, handle it and reset.
 328     if (_messageSize >= _totalMessageSize) {
 329       [_delegate messageQueue:self didReceiveMessage:[_message autorelease]];
 330       _message = nil;
 331
 332       // Process any outgoing messages.
 333       [self dequeueAndSend];
 334     }
 335   }
 336 }
 337
 338 - (void)listenSocket:(CFSocketRef)socket acceptedSocket:(CFSocketNativeHandle)child {
 339   if (socket != _socket) {
 340     // TODO: error
 341     return;
 342   }
 343
 344   _child = child;
 345
 346   // Create the streams on the socket.
 347   CFStreamCreatePairWithSocket(kCFAllocatorDefault,
 348                                _child,  // Socket handle.
 349                                &_readStream,  // Read stream in-pointer.
 350                                &_writeStream);  // Write stream in-pointer.
 351
 352   // Create struct to register callbacks for the stream.
 353   CFStreamClientContext context = { 0 };
 354   context.info = self;
 355
 356   // Set the client of the read stream.
 357   CFOptionFlags readFlags = kCFStreamEventOpenCompleted |
 358                             kCFStreamEventHasBytesAvailable |
 359                             kCFStreamEventErrorOccurred |
 360                             kCFStreamEventEndEncountered;
 361   if (CFReadStreamSetClient(_readStream, readFlags, &MessageQueueReadEvent, &context))
 362     // Schedule in run loop to do asynchronous communication with the engine.
 363     CFReadStreamScheduleWithRunLoop(_readStream, [_runLoop getCFRunLoop], kCFRunLoopCommonModes);
 364   else
 365     return;
 366
 367   // Open the stream now that it's scheduled on the run loop.
 368   if (!CFReadStreamOpen(_readStream)) {
 369     [self reportError:CFReadStreamCopyError(_readStream)];
 370     return;
 371   }
 372
 373   // Set the client of the write stream.
 374   CFOptionFlags writeFlags = kCFStreamEventOpenCompleted |
 375                              kCFStreamEventCanAcceptBytes |
 376                              kCFStreamEventErrorOccurred |
 377                              kCFStreamEventEndEncountered;
 378   if (CFWriteStreamSetClient(_writeStream, writeFlags, &MessageQueueWriteEvent, &context))
 379     // Schedule it in the run loop to receive error information.
 380     CFWriteStreamScheduleWithRunLoop(_writeStream, [_runLoop getCFRunLoop], kCFRunLoopCommonModes);
 381   else
 382     return;
 383
 384   // Open the write stream.
 385   if (!CFWriteStreamOpen(_writeStream)) {
 386     [self reportError:CFWriteStreamCopyError(_writeStream)];
 387     return;
 388   }
 389
 390   _connected = YES;
 391   [_delegate messageQueueDidConnect:self];
 392
 393   CFSocketInvalidate(_socket);
 394   CFRelease(_socket);
 395   _socket = NULL;
 396 }
 397
 398 - (void)reportError:(CFErrorRef)error
 399 {
 400   [_delegate messageQueue:self error:(NSError*)error];
 401   CFRelease(error);
 402 }
 403
 404 - (void)readStream:(CFReadStreamRef)stream handleEvent:(CFStreamEventType)event
 405 {
 406   assert(stream == _readStream);
 407   switch (event)
 408   {
 409     case kCFStreamEventHasBytesAvailable:
 410       [self readMessageFromStream];
 411       break;
 412
 413     case kCFStreamEventErrorOccurred:
 414       [self reportError:CFReadStreamCopyError(stream)];
 415       [self stopRunLoop];
 416       break;
 417
 418     case kCFStreamEventEndEncountered:
 419       [self stopRunLoop];
 420       break;
 421
 422     default:
 423       // TODO: error
 424       break;
 425   };
 426 }
 427
 428 - (void)writeStream:(CFWriteStreamRef)stream handleEvent:(CFStreamEventType)event
 429 {
 430   assert(stream == _writeStream);
 431   switch (event) {
 432     case kCFStreamEventCanAcceptBytes:
 433       [self dequeueAndSend];
 434       break;
 435
 436     case kCFStreamEventErrorOccurred:
 437       [self reportError:CFWriteStreamCopyError(stream)];
 438       [self stopRunLoop];
 439       break;
 440
 441     case kCFStreamEventEndEncountered:
 442       [self stopRunLoop];
 443       break;
 444
 445     default:
 446       // TODO: error
 447       break;
 448   }
 449 }
 450
 451 @end