]> src.bluestatic.org Git - macgdbp.git/blob - Source/MessageQueue.m
Do not double-percent-escape the filename.
[macgdbp.git] / 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