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