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     [self scheduleListenSocket];
 144
 145     // Use CFRunLoop instead of NSRunLoop because the latter has no programmatic
 146     // stop routine.
 147     CFRunLoopRun();
 148
 149     _thread = nil;
 150     _runLoop = nil;
 151   }
 152 }
 153
 154 - (void)scheduleListenSocket {
 155   // Create the address structure.
 156   struct sockaddr_in address;
 157   memset(&address, 0, sizeof(address));
 158   address.sin_len = sizeof(address);
 159   address.sin_family = AF_INET;
 160   address.sin_port = htons(_port);
 161   address.sin_addr.s_addr = htonl(INADDR_ANY);
 162
 163   // Create the socket signature.
 164   CFSocketSignature signature;
 165   signature.protocolFamily = PF_INET;
 166   signature.socketType = SOCK_STREAM;
 167   signature.protocol = IPPROTO_TCP;
 168   signature.address = (CFDataRef)[NSData dataWithBytes:&address length:sizeof(address)];
 169
 170   CFSocketContext context = { 0 };
 171   context.info = self;
 172
 173   do {
 174     _socket =
 175         CFSocketCreateWithSocketSignature(kCFAllocatorDefault,
 176                                           &signature,  // Socket signature.
 177                                           kCFSocketAcceptCallBack,  // Callback types.
 178                                           &MessageQueueSocketAccept,  // Callback function.
 179                                           &context);  // Context to pass to callout.
 180     if (!_socket) {
 181       //[connection_ errorEncountered:@"Could not open socket."];
 182       sleep(1);
 183     }
 184   } while (!_socket);
 185
 186   // Allow old, yet-to-be recycled sockets to be reused.
 187   int yes = 1;
 188   setsockopt(CFSocketGetNative(_socket), SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int));
 189   setsockopt(CFSocketGetNative(_socket), SOL_SOCKET, SO_REUSEPORT, &yes, sizeof(int));
 190
 191   // Schedule the socket on the run loop.
 192   CFRunLoopSourceRef source = CFSocketCreateRunLoopSource(kCFAllocatorDefault, _socket, 0);
 193   CFRunLoopAddSource([_runLoop getCFRunLoop], source, kCFRunLoopCommonModes);
 194   CFRelease(source);
 195 }
 196
 197 - (void)disconnectClient {
 198   if (_readStream) {
 199     CFReadStreamUnscheduleFromRunLoop(_readStream, [_runLoop getCFRunLoop], kCFRunLoopCommonModes);
 200     CFReadStreamClose(_readStream);
 201     CFRelease(_readStream);
 202     _readStream = NULL;
 203   }
 204
 205   if (_writeStream) {
 206     CFWriteStreamUnscheduleFromRunLoop(_writeStream, [_runLoop getCFRunLoop], kCFRunLoopCommonModes);
 207     CFWriteStreamClose(_writeStream);
 208     CFRelease(_writeStream);
 209     _writeStream = NULL;
 210   }
 211
 212   if (_child) {
 213     close(_child);
 214     _child = NULL;
 215   }
 216
 217   _connected = NO;
 218   [_delegate messageQueueDidDisconnect:self];
 219 }
 220
 221 - (void)stopRunLoop {
 222   [self disconnectClient];
 223   CFRunLoopStop([_runLoop getCFRunLoop]);
 224 }
 225
 226 - (void)enqueueMessage:(NSString*)message {
 227   [_queue addObject:message];
 228   [self dequeueAndSend];
 229 }
 230
 231 - (void)dequeueAndSend {
 232   if (![_queue count])
 233     return;
 234
 235   if (!CFWriteStreamCanAcceptBytes(_writeStream))
 236     return;
 237
 238   NSString* message = [_queue objectAtIndex:0];
 239   [self performSend:message];
 240   [_queue removeObjectAtIndex:0];
 241 }
 242
 243 - (void)performSend:(NSString*)message {
 244   // TODO: May need to negotiate with the server as to the string encoding.
 245   const NSStringEncoding kEncoding = NSUTF8StringEncoding;
 246   // Add space for the NUL byte.
 247   NSUInteger maxBufferSize = [message maximumLengthOfBytesUsingEncoding:kEncoding] + 1;
 248
 249   UInt8* buffer = malloc(maxBufferSize);
 250   bzero(buffer, maxBufferSize);
 251
 252   NSUInteger bufferSize = 0;
 253   if (![message getBytes:buffer
 254                maxLength:maxBufferSize
 255               usedLength:&bufferSize
 256                 encoding:kEncoding
 257                  options:0
 258                    range:NSMakeRange(0, [message length])
 259           remainingRange:NULL]) {
 260     free(buffer);
 261     return;
 262   }
 263
 264   // Include a NUL byte.
 265   ++bufferSize;
 266
 267   // Write the packet out, and spin in a busy wait loop if the stream is not ready. This
 268   // method is only ever called in response to a stream ready event.
 269   NSUInteger totalWritten = 0;
 270   while (totalWritten < bufferSize) {
 271     CFIndex bytesWritten = CFWriteStreamWrite(_writeStream, buffer + totalWritten, bufferSize - totalWritten);
 272     if (bytesWritten < 0) {
 273       [self reportError:CFWriteStreamCopyError(_writeStream)];
 274       break;
 275     }
 276     totalWritten += bytesWritten;
 277   }
 278
 279   [_delegate messageQueue:self didSendMessage:message];
 280
 281   free(buffer);
 282 }
 283
 284 - (void)readMessageFromStream {
 285   const NSUInteger kBufferSize = 1024;
 286   UInt8 buffer[kBufferSize];
 287   CFIndex bufferOffset = 0;  // Starting point in |buffer| to work with.
 288   CFIndex bytesRead = CFReadStreamRead(_readStream, buffer, kBufferSize);
 289   const char* charBuffer = (const char*)buffer;
 290
 291   // The read loop works by going through the buffer until all the bytes have
 292   // been processed.
 293   while (bufferOffset < bytesRead) {
 294     // Find the NUL separator, or the end of the string.
 295     NSUInteger partLength = 0;
 296     for (CFIndex i = bufferOffset; i < bytesRead && charBuffer[i] != '\0'; ++i, ++partLength) ;
 297
 298     // If there is not a current packet, set some state.
 299     if (!_message) {
 300       // Read the message header: the size.  This will be |partLength| bytes.
 301       _totalMessageSize = atoi(charBuffer + bufferOffset);
 302       _messageSize = 0;
 303       _message = [[NSMutableString alloc] initWithCapacity:_totalMessageSize];
 304       bufferOffset += partLength + 1;  // Pass over the NUL byte.
 305       continue;  // Spin the loop to begin reading actual data.
 306     }
 307
 308     // Substring the byte stream and append it to the packet string.
 309     CFStringRef bufferString = CFStringCreateWithBytes(kCFAllocatorDefault,
 310                                                        buffer + bufferOffset,  // Byte pointer, offset by start index.
 311                                                        partLength,  // Length.
 312                                                        kCFStringEncodingUTF8,
 313                                                        true);
 314     [_message appendString:(NSString*)bufferString];
 315     CFRelease(bufferString);
 316
 317     // Advance counters.
 318     _messageSize += partLength;
 319     bufferOffset += partLength + 1;
 320
 321     // If this read finished the packet, handle it and reset.
 322     if (_messageSize >= _totalMessageSize) {
 323       [_delegate messageQueue:self didReceiveMessage:[_message autorelease]];
 324       _message = nil;
 325
 326       // Process any outgoing messages.
 327       [self dequeueAndSend];
 328     }
 329   }
 330 }
 331
 332 - (void)listenSocket:(CFSocketRef)socket acceptedSocket:(CFSocketNativeHandle)child {
 333   if (socket != _socket) {
 334     // TODO: error
 335     return;
 336   }
 337
 338   _child = child;
 339
 340   // Create the streams on the socket.
 341   CFStreamCreatePairWithSocket(kCFAllocatorDefault,
 342                                _child,  // Socket handle.
 343                                &_readStream,  // Read stream in-pointer.
 344                                &_writeStream);  // Write stream in-pointer.
 345
 346   // Create struct to register callbacks for the stream.
 347   CFStreamClientContext context = { 0 };
 348   context.info = self;
 349
 350   // Set the client of the read stream.
 351   CFOptionFlags readFlags = kCFStreamEventOpenCompleted |
 352                             kCFStreamEventHasBytesAvailable |
 353                             kCFStreamEventErrorOccurred |
 354                             kCFStreamEventEndEncountered;
 355   if (CFReadStreamSetClient(_readStream, readFlags, &MessageQueueReadEvent, &context))
 356     // Schedule in run loop to do asynchronous communication with the engine.
 357     CFReadStreamScheduleWithRunLoop(_readStream, [_runLoop getCFRunLoop], kCFRunLoopCommonModes);
 358   else
 359     return;
 360
 361   // Open the stream now that it's scheduled on the run loop.
 362   if (!CFReadStreamOpen(_readStream)) {
 363     [self reportError:CFReadStreamCopyError(_readStream)];
 364     return;
 365   }
 366
 367   // Set the client of the write stream.
 368   CFOptionFlags writeFlags = kCFStreamEventOpenCompleted |
 369   kCFStreamEventCanAcceptBytes |
 370   kCFStreamEventErrorOccurred |
 371   kCFStreamEventEndEncountered;
 372   if (CFWriteStreamSetClient(_writeStream, writeFlags, &MessageQueueWriteEvent, &context))
 373     // Schedule it in the run loop to receive error information.
 374     CFWriteStreamScheduleWithRunLoop(_writeStream, [_runLoop getCFRunLoop], kCFRunLoopCommonModes);
 375   else
 376     return;
 377
 378   // Open the write stream.
 379   if (!CFWriteStreamOpen(_writeStream)) {
 380     [self reportError:CFWriteStreamCopyError(_writeStream)];
 381     return;
 382   }
 383
 384   _connected = YES;
 385   [_delegate messageQueueDidConnect:self];
 386
 387   CFSocketInvalidate(_socket);
 388   CFRelease(_socket);
 389   _socket = NULL;
 390 }
 391
 392 - (void)reportError:(CFErrorRef)error
 393 {
 394   [_delegate messageQueue:self error:(NSError*)error];
 395   CFRelease(error);
 396 }
 397
 398 - (void)readStream:(CFReadStreamRef)stream handleEvent:(CFStreamEventType)event
 399 {
 400   assert(stream == _readStream);
 401   switch (event)
 402   {
 403     case kCFStreamEventHasBytesAvailable:
 404       [self readMessageFromStream];
 405       break;
 406
 407     case kCFStreamEventErrorOccurred:
 408       [self reportError:CFReadStreamCopyError(stream)];
 409       [self stopRunLoop];
 410       break;
 411
 412     case kCFStreamEventEndEncountered:
 413       [self stopRunLoop];
 414       break;
 415
 416     default:
 417       // TODO: error
 418       break;
 419   };
 420 }
 421
 422 - (void)writeStream:(CFWriteStreamRef)stream handleEvent:(CFStreamEventType)event
 423 {
 424   assert(stream == _writeStream);
 425   switch (event) {
 426     case kCFStreamEventCanAcceptBytes:
 427       [self dequeueAndSend];
 428       break;
 429
 430     case kCFStreamEventErrorOccurred:
 431       [self reportError:CFWriteStreamCopyError(stream)];
 432       [self stopRunLoop];
 433       break;
 434
 435     case kCFStreamEventEndEncountered:
 436       [self stopRunLoop];
 437       break;
 438
 439     default:
 440       // TODO: error
 441       break;
 442   }
 443 }
 444
 445 @end