AFMultipartBodyStream 不支持 scheduleInRunLoop 机制的原因是什么？

Question

虽然AFNetworking lib 的AFMultipartBodyStream 是NSStream 的子类，符合NSStreamDelegate 协议，但它不能像常规NSStream 的标准方式那样处理。即，AFMultipartBodyStream 无法使用流事件处理。我查看了AFMultipartBodyStream的代码，发现它故意禁用了NSInputStream抽象类的scheduleInRunLoop方法：

- (void)scheduleInRunLoop:(__unused NSRunLoop *)aRunLoop
              forMode:(__unused NSString *)mode
{}

- (void)removeFromRunLoop:(__unused NSRunLoop *)aRunLoop
              forMode:(__unused NSString *)mode
{}

有什么具体原因吗？是不是让它支持标准的流事件机制，使得流数据的处理可以通过stream:handleEvent:事件处理器异步完成？

Answer 1

在研究了AFMultipartBodyStream的实现之后，我注意到目前的实现方式无法支持常规流IO处理的异步方式。然后我增强了 AFMultipartBodyStream 以提供一个与内部多部分数据结构连接的流，因此这个 AFMultipartBodyStream 的持有者可以将多部分数据作为可以在运行循环中调度的常规流来处理。下面的代码 sn-p 显示了主要思想：

-(NSInputStream *)inputStream {
    // If _inputStream has not been connected with HTTPBodyParts data, establish the connection
    if (!_inputStream) {
        NSParameterAssert([self.HTTPBodyParts count] != 0);
        CFReadStreamRef readStream;
        CFWriteStreamRef writeStream;
        CFIndex bufferSize = self.contentLength;
        CFStreamCreateBoundPair(NULL, &readStream, &writeStream, bufferSize);
        _inputStream = (__bridge_transfer NSInputStream *)readStream;
        _outputStream = (__bridge_transfer NSOutputStream *)writeStream;
        [_outputStream setDelegate:self];

        dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_HIGH, 0), ^{
            NSLog(@"\n====in async block of inputStream....====\n");

            [self->_outputStream scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
            [self open];
            [self->_outputStream open];
            NSInteger totalBytesSent = self.contentLength;
            while (totalBytesSent > 0 && [self->_outputStream hasSpaceAvailable]) {
                uint8_t buffer[1024];
                NSInteger bytesRead = [self read:buffer maxLength:1024];
                totalBytesSent -= bytesRead;
                NSLog(@"\n====buffer read (%ld): [%s]====\n", (long)bytesRead, buffer);
                if (self.streamError || bytesRead < 0) {
                    break;
                }

                NSInteger bytesWritten = [self->_outputStream write:buffer maxLength:(NSUInteger)bytesRead];
                if (self->_outputStream.streamError || bytesWritten < 0) {
                    NSLog(@"\n====Socket write failed[%@]====\n", self->_outputStream.streamError);
                    break;
                }

                if (bytesRead == 0 && bytesWritten == 0) {
                    break;
                }
            }

            [self->_outputStream close];
            [self->_outputStream removeFromRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
        });
    }

    return _inputStream;
}




// Added
- (void)scheduleInRunLoop:(__unused NSRunLoop *)aRunLoop
                  forMode:(__unused NSString *)mode
{
    // Setup the input stream for body stream data consuming
    NSInputStream *inputStream = [self inputStream];
    NSParameterAssert(inputStream == self.inputStream);
    [inputStream setDelegate:self.delegate_];
    [inputStream scheduleInRunLoop:aRunLoop forMode:mode];
    [inputStream open];
}
// Added
- (void)removeFromRunLoop:(__unused NSRunLoop *)aRunLoop
                  forMode:(__unused NSString *)mode
{
    if (_inputStream) {
        [_inputStream setDelegate:[self delegate]];
        [_inputStream removeFromRunLoop:aRunLoop forMode:mode];
        [_inputStream close];
    }
}