Overview
Comment: | Add -[asyncReadIntoBuffer:exactLength:block:]. |
---|---|
Downloads: | Tarball | ZIP archive | SQL archive |
Timelines: | family | ancestors | descendants | both | trunk |
Files: | files | file ages | folders |
SHA3-256: |
be6adc6e06fe6c1c275c20629a972bf9 |
User & Date: | js on 2012-09-17 15:48:52 |
Other Links: | manifest | tags |
Context
2012-09-19
| ||
23:02 | Make async I/O possible without blocks. check-in: 4d9a641d87 user: js tags: trunk | |
2012-09-17
| ||
15:48 | Add -[asyncReadIntoBuffer:exactLength:block:]. check-in: be6adc6e06 user: js tags: trunk | |
10:18 | Add -[performSelectorOnMainThread:waitUntilDone:]. check-in: e959c99f85 user: js tags: trunk | |
Changes
Modified src/OFRunLoop.h from [f3912ea6ee] to [e532eb9c3d].
︙ | ︙ | |||
51 52 53 54 55 56 57 58 59 60 61 62 63 64 | + (OFRunLoop*)currentRunLoop; + (void)OF_setMainRunLoop; #ifdef OF_HAVE_BLOCKS + (void)OF_addAsyncReadForStream: (OFStream*)stream buffer: (void*)buffer length: (size_t)length block: (of_stream_async_read_block_t)block; + (void)OF_addAsyncReadLineForStream: (OFStream*)stream encoding: (of_string_encoding_t)encoding block: (of_stream_async_read_line_block_t)block; + (void)OF_addAsyncAcceptForTCPSocket: (OFTCPSocket*)socket block: (of_tcpsocket_async_accept_block_t)block; #endif | > > > > | 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | + (OFRunLoop*)currentRunLoop; + (void)OF_setMainRunLoop; #ifdef OF_HAVE_BLOCKS + (void)OF_addAsyncReadForStream: (OFStream*)stream buffer: (void*)buffer length: (size_t)length block: (of_stream_async_read_block_t)block; + (void)OF_addAsyncReadForStream: (OFStream*)stream buffer: (void*)buffer exactLength: (size_t)length block: (of_stream_async_read_block_t)block; + (void)OF_addAsyncReadLineForStream: (OFStream*)stream encoding: (of_string_encoding_t)encoding block: (of_stream_async_read_line_block_t)block; + (void)OF_addAsyncAcceptForTCPSocket: (OFTCPSocket*)socket block: (of_tcpsocket_async_accept_block_t)block; #endif |
︙ | ︙ |
Modified src/OFRunLoop.m from [4507c3903b] to [2fb0c01789].
︙ | ︙ | |||
29 30 31 32 33 34 35 36 37 38 39 40 | #import "macros.h" static OFRunLoop *mainRunLoop = nil; #ifdef OF_HAVE_BLOCKS @interface OFRunLoop_ReadQueueItem: OFObject { void *buffer; size_t length; of_stream_async_read_block_t block; } | > > > > > | | | > > < < < > < < > > > | > | > > > < < < < | 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | #import "macros.h" static OFRunLoop *mainRunLoop = nil; #ifdef OF_HAVE_BLOCKS @interface OFRunLoop_ReadQueueItem: OFObject { @public void *buffer; size_t length; of_stream_async_read_block_t block; } @end @interface OFRunLoop_ExactReadQueueItem: OFObject { @public void *buffer; size_t exactLength, readLength; of_stream_async_read_block_t block; } @end @interface OFRunLoop_ReadLineQueueItem: OFObject { @public of_stream_async_read_line_block_t block; of_string_encoding_t encoding; } @end @interface OFRunLoop_AcceptQueueItem: OFObject { @public of_tcpsocket_async_accept_block_t block; } @end @implementation OFRunLoop_ReadQueueItem - (void)dealloc { [block release]; [super dealloc]; } @end @implementation OFRunLoop_ExactReadQueueItem - (void)dealloc { [block release]; [super dealloc]; } @end @implementation OFRunLoop_ReadLineQueueItem - (void)dealloc { [block release]; [super dealloc]; } @end @implementation OFRunLoop_AcceptQueueItem - (void)dealloc { [block release]; [super dealloc]; } @end |
︙ | ︙ | |||
111 112 113 114 115 116 117 | void *pool = objc_autoreleasePoolPush(); mainRunLoop = [[self currentRunLoop] retain]; objc_autoreleasePoolPop(pool); } #ifdef OF_HAVE_BLOCKS + (void)OF_addAsyncReadForStream: (OFStream*)stream | | | | > > > > | > > | > > > > > > > > > | > > > > > > > > > > > > > | 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 | void *pool = objc_autoreleasePoolPush(); mainRunLoop = [[self currentRunLoop] retain]; objc_autoreleasePoolPop(pool); } #ifdef OF_HAVE_BLOCKS + (void)OF_addAsyncReadForStream: (OFStream*)stream buffer: (void*)buffer length: (size_t)length block: (of_stream_async_read_block_t)block { void *pool = objc_autoreleasePoolPush(); OFRunLoop *runLoop = [self currentRunLoop]; OFList *queue = [runLoop->readQueues objectForKey: stream]; OFRunLoop_ReadQueueItem *queueItem; if (queue == nil) { queue = [OFList list]; [runLoop->readQueues setObject: queue forKey: stream]; } if ([queue count] == 0) [runLoop->streamObserver addStreamForReading: stream]; queueItem = [[[OFRunLoop_ReadQueueItem alloc] init] autorelease]; queueItem->buffer = buffer; queueItem->length = length; queueItem->block = [block copy]; [queue appendObject: queueItem]; objc_autoreleasePoolPop(pool); } + (void)OF_addAsyncReadForStream: (OFStream*)stream buffer: (void*)buffer exactLength: (size_t)exactLength block: (of_stream_async_read_block_t)block { void *pool = objc_autoreleasePoolPush(); OFRunLoop *runLoop = [self currentRunLoop]; OFList *queue = [runLoop->readQueues objectForKey: stream]; OFRunLoop_ExactReadQueueItem *queueItem; if (queue == nil) { queue = [OFList list]; [runLoop->readQueues setObject: queue forKey: stream]; } if ([queue count] == 0) [runLoop->streamObserver addStreamForReading: stream]; queueItem = [[[OFRunLoop_ExactReadQueueItem alloc] init] autorelease]; queueItem->buffer = buffer; queueItem->exactLength = exactLength; queueItem->block = [block copy]; [queue appendObject: queueItem]; objc_autoreleasePoolPop(pool); } + (void)OF_addAsyncReadLineForStream: (OFStream*)stream encoding: (of_string_encoding_t)encoding |
︙ | ︙ | |||
157 158 159 160 161 162 163 | forKey: stream]; } if ([queue count] == 0) [runLoop->streamObserver addStreamForReading: stream]; queueItem = [[[OFRunLoop_ReadLineQueueItem alloc] init] autorelease]; | | | | 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 | forKey: stream]; } if ([queue count] == 0) [runLoop->streamObserver addStreamForReading: stream]; queueItem = [[[OFRunLoop_ReadLineQueueItem alloc] init] autorelease]; queueItem->block = [block copy]; queueItem->encoding = encoding; [queue appendObject: queueItem]; objc_autoreleasePoolPop(pool); } + (void)OF_addAsyncAcceptForTCPSocket: (OFTCPSocket*)socket block: (of_tcpsocket_async_accept_block_t)block |
︙ | ︙ | |||
182 183 184 185 186 187 188 | forKey: socket]; } if ([queue count] == 0) [runLoop->streamObserver addStreamForReading: socket]; queueItem = [[[OFRunLoop_AcceptQueueItem alloc] init] autorelease]; | | | 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 | forKey: socket]; } if ([queue count] == 0) [runLoop->streamObserver addStreamForReading: socket]; queueItem = [[[OFRunLoop_AcceptQueueItem alloc] init] autorelease]; queueItem->block = [block copy]; [queue appendObject: queueItem]; objc_autoreleasePoolPop(pool); } #endif - init |
︙ | ︙ | |||
238 239 240 241 242 243 244 | OF_ENSURE(queue != nil); listObject = [queue firstListObject]; if ([listObject->object isKindOfClass: [OFRunLoop_ReadQueueItem class]]) { OFRunLoop_ReadQueueItem *queueItem = listObject->object; | < | | | > > > > > > > > > > > > > > > > > > > > > > > > > | | | | 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 | OF_ENSURE(queue != nil); listObject = [queue firstListObject]; if ([listObject->object isKindOfClass: [OFRunLoop_ReadQueueItem class]]) { OFRunLoop_ReadQueueItem *queueItem = listObject->object; size_t length = [stream readIntoBuffer: queueItem->buffer length: queueItem->length]; if (!queueItem->block(stream, queueItem->buffer, length)) { [queue removeListObject: listObject]; if ([queue count] == 0) { [streamObserver removeStreamForReading: stream]; [readQueues removeObjectForKey: stream]; } } } else if ([listObject->object isKindOfClass: [OFRunLoop_ExactReadQueueItem class]]) { OFRunLoop_ExactReadQueueItem *queueItem = listObject->object; size_t length = [stream readIntoBuffer: (char*)queueItem->buffer + queueItem->readLength length: queueItem->exactLength - queueItem->readLength]; queueItem->readLength += length; if (queueItem->readLength == queueItem->exactLength || [stream isAtEndOfStream]) { if (queueItem->block(stream, queueItem->buffer, queueItem->readLength)) queueItem->readLength = 0; else { [queue removeListObject: listObject]; if ([queue count] == 0) { [streamObserver removeStreamForReading: stream]; [readQueues removeObjectForKey: stream]; } } } } else if ([listObject->object isKindOfClass: [OFRunLoop_ReadLineQueueItem class]]) { OFRunLoop_ReadLineQueueItem *queueItem = listObject->object; OFString *line; line = [stream tryReadLineWithEncoding: queueItem->encoding]; if (line != nil || [stream isAtEndOfStream]) { if (!queueItem->block(stream, line)) { [queue removeListObject: listObject]; if ([queue count] == 0) { [streamObserver removeStreamForReading: stream]; [readQueues removeObjectForKey: stream]; } } } } else if ([listObject->object isKindOfClass: [OFRunLoop_AcceptQueueItem class]]) { OFRunLoop_AcceptQueueItem *queueItem = listObject->object; OFTCPSocket *newSocket = [(OFTCPSocket*)stream accept]; if (!queueItem->block((OFTCPSocket*)stream, newSocket)) { [queue removeListObject: listObject]; if ([queue count] == 0) { [streamObserver removeStreamForReading: stream]; [readQueues removeObjectForKey: stream]; } } |
︙ | ︙ |
Modified src/OFStream.h from [67eb3e5d0e] to [428dff50a5].
︙ | ︙ | |||
74 75 76 77 78 79 80 | - (BOOL)isAtEndOfStream; /** * \brief Reads <i>at most</i> size bytes from the stream into a buffer. * * On network streams, this might read less than the specified number of bytes. * If you want to read exactly the specified number of bytes, use | | > > < < < < < < < < < < < < < < < < < < < < < < < < | > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > | 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 | - (BOOL)isAtEndOfStream; /** * \brief Reads <i>at most</i> size bytes from the stream into a buffer. * * On network streams, this might read less than the specified number of bytes. * If you want to read exactly the specified number of bytes, use * -readIntoBuffer:exactLength:. Note that a read can even return 0 bytes - * this does not necessarily mean that the stream ended, so you still need to * check isAtEndOfStream. * * \param buffer The buffer into which the data is read * \param length The length of the data that should be read at most. * The buffer <i>must</i> be at least this big! * \return The number of bytes read */ - (size_t)readIntoBuffer: (void*)buffer length: (size_t)size; /** * \brief Reads exactly the specified length bytes from the stream into a * buffer. * * Unlike readIntoBuffer:length:, this method does not return when less than the * specified length has been read - instead, it waits until it got exactly the * specified length. * * \warning Only call this when you know that specified amount of data is * available! Otherwise you will get an exception! * * \param buffer The buffer into which the data is read * \param length The length of the data that should be read. * The buffer <i>must</i> be <i>exactly</i> this big! */ - (void)readIntoBuffer: (void*)buffer exactLength: (size_t)length; #ifdef OF_HAVE_BLOCKS /** * \brief Asyncronously reads <i>at most</i> size bytes from the stream into a * buffer. * * On network streams, this might read less than the specified number of bytes. * If you want to read exactly the specified number of bytes, use * asyncReadIntoBuffer:exactLength:block:. Note that a read can even return 0 * bytes - this does not necessarily mean that the stream ended, so you still * need to check isAtEndOfStream. * * \param buffer The buffer into which the data is read. * The buffer must not be free'd before the async read completed! * \param length The length of the data that should be read at most. * The buffer <i>must</i> be at least this big! * \param block The block to call when the data has been received. * If the block returns YES, it will be called again with the same * buffer and maximum length when more data has been received. If * you want the next block in the queue to handle the data * received next, you need to return NO from the block. */ - (void)asyncReadIntoBuffer: (void*)buffer length: (size_t)length block: (of_stream_async_read_block_t)block; /** * \brief Asyncronously reads exactly the specified length bytes from the * stream into a buffer. * * Unlike asyncReadIntoBuffer:length:block, this method does not invoke the * block when less than the specified length has been read - instead, it waits * until it got exactly the specified length or the stream has ended. * * \param buffer The buffer into which the data is read * \param length The length of the data that should be read. * The buffer <i>must</i> be <i>exactly</i> this big! * \param block The block to call when the data has been received. * If the block returns YES, it will be called again with the same * buffer and exact length when more data has been received. If * you want the next block in the queue to handle the data * received next, you need to return NO from the block. */ - (void)asyncReadIntoBuffer: (void*)buffer exactLength: (size_t)length block: (of_stream_async_read_block_t)block; #endif /** * \brief Reads a uint8_t from the stream. * * \warning Only call this when you know that enough data is available! * Otherwise you will get an exception! * |
︙ | ︙ |
Modified src/OFStream.m from [92797b60e4] to [befd533abe].
︙ | ︙ | |||
130 131 132 133 134 135 136 | cache = tmp; cacheLength -= length; return length; } } | | > > > > > > > > > > > | | > < | < | | > > | 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 | cache = tmp; cacheLength -= length; return length; } } - (void)readIntoBuffer: (void*)buffer exactLength: (size_t)length { size_t readLength = 0; while (readLength < length) readLength += [self readIntoBuffer: (char*)buffer + readLength length: length - readLength]; } #ifdef OF_HAVE_BLOCKS - (void)asyncReadIntoBuffer: (void*)buffer length: (size_t)length block: (of_stream_async_read_block_t)block { [OFRunLoop OF_addAsyncReadForStream: self buffer: buffer length: length block: block]; } - (void)asyncReadIntoBuffer: (void*)buffer exactLength: (size_t)length block: (of_stream_async_read_block_t)block { [OFRunLoop OF_addAsyncReadForStream: self buffer: buffer exactLength: length block: block]; } #endif - (uint8_t)readInt8 { uint8_t ret; [self readIntoBuffer: (char*)&ret exactLength: 1]; |
︙ | ︙ |
Modified src/OFStreamObserver.m from [4eb84860f4] to [62e60a6cd0].
︙ | ︙ | |||
380 381 382 383 384 385 386 | { OFStream **objects = [readStreams objects]; size_t i, count = [readStreams count]; BOOL foundInCache = NO; for (i = 0; i < count; i++) { | < | 380 381 382 383 384 385 386 387 388 389 390 391 392 393 | { OFStream **objects = [readStreams objects]; size_t i, count = [readStreams count]; BOOL foundInCache = NO; for (i = 0; i < count; i++) { if ([objects[i] pendingBytes] > 0 && ![objects[i] OF_isWaitingForDelimiter]) { void *pool = objc_autoreleasePoolPush(); [delegate streamIsReadyForReading: objects[i]]; foundInCache = YES; objc_autoreleasePoolPop(pool); } |
︙ | ︙ |