ios - 如何实现超时/等待NSStream有效地使方法同步

Question

我有蓝牙连接附件的输入流和输出流

我要实现以下目标:

将数据写入outputStream
等待，直到在inputStream上接收到数据，或者直到10秒钟过去
如果inputStream数据到达，则返回数据
否则返回nil

我试图这样实现:

- (APDUResponse *)sendCommandAndWaitForResponse:(NSData *)request {
  APDUResponse * result;
  if (!deviceIsBusy && request != Nil) {
    deviceIsBusy = YES;
    timedOut = NO;
    responseReceived = NO;
    if ([[mySes outputStream] hasSpaceAvailable]) {
      [NSThread detachNewThreadSelector:@selector(startTimeout) toTarget:self withObject:nil];
      [[mySes outputStream] write:[request bytes] maxLength:[request length]];
      while (!timedOut && !responseReceived) {
        sleep(2);
        NSLog(@"tick");
      }
      if (responseReceived && response !=nil) {
        result = response;
        response = nil;
      }
      [myTimer invalidate];
      myTimer = nil;
    }
  }
  deviceIsBusy = NO;
  return result;
}

- (void) startTimeout {
  NSLog(@"start Timeout");
  myTimer = [NSTimer timerWithTimeInterval:10.0 target:self selector:@selector(timerFireMethod:) userInfo:nil repeats:YES];
  [[NSRunLoop currentRunLoop] addTimer:myTimer forMode:NSRunLoopCommonModes];
}

- (void)timerFireMethod:(NSTimer *)timer {
  NSLog(@"fired");
  timedOut = YES;
}

- (void)stream:(NSStream*)stream handleEvent:(NSStreamEvent)streamEvent
{
  switch (streamEvent)
  {
    case NSStreamEventHasBytesAvailable:
      // Process the incoming stream data.
      if(stream == [mySes inputStream])
      {
        uint8_t buf[1024];
        unsigned int len = 0;
        len = [[mySes inputStream] read:buf maxLength:1024];
        if(len) {
          _data = [[NSMutableData alloc] init];
          [_data appendBytes:(const void *)buf length:len];
          NSLog(@"Response: %@", [_data description]);
          response = [[APDUResponse alloc] initWithData:_data];
          responseReceived = YES;
        } else {
          NSLog(@"no buffer!");
        }
      }
      break;
     ... //code not relevant 
  }
}

因此，理论上是让NSTimer在单独的线程上运行，该线程将在触发时设置一个 bool 值，然后如果接收到数据，则使handleEvent委托(delegate)方法设置另一个 bool 值。
在该方法中，我们有一个带有 sleep 的while循环，当设置了其中一个 bool 值时，该循环将停止。

我遇到的问题是在“超时情况”下我的timerFireMethod没有被调用。我的直觉是我实际上没有在单独的线程上正确设置计时器。

谁能看到这里出了什么问题，或者为上述要求提出了更好的实现方案？

Answer 1

而是对固有的异步问题强加不合适的同步方法，而应使方法sendCommandAndWaitForResponse异步。

可以将“流写入”任务包装到异步操作/任务/方法中。例如，您可以使用以下接口(interface)以NSOperation的并发子类结尾:

typedef void (^DataToStreamCopier_completion_t)(id result);

@interface DataToStreamCopier : NSOperation

- (id) initWithData:(NSData*)sourceData
  destinationStream:(NSOutputStream*)destinationStream
         completion:(DataToStreamCopier_completion_t)completionHandler;

@property (nonatomic) NSThread* workerThread;
@property (nonatomic, copy) NSString* runLoopMode;
@property (atomic, readonly) long long totalBytesCopied;


// NSOperation
- (void) start;
- (void) cancel;
@property (nonatomic, readonly) BOOL isCancelled;
@property (nonatomic, readonly) BOOL isExecuting;
@property (nonatomic, readonly) BOOL isFinished;

@end

您可以使用 cancel方法实现“超时”功能。

您的方法 sendCommandAndWaitForResponse:与完成处理程序变为异步:

- (void)sendCommand:(NSData *)request 
         completion:(DataToStreamCopier_completion_t)completionHandler
{
    DataToStreamCopier* op = [DataToStreamCopier initWithData:request 
                                            destinationStream:self.outputStream 
                                                   completion:completionHandler];
   [op start];

   // setup timeout with block:  ^{ [op cancel]; }
   ...
}

用法:

[self sendCommand:request completion:^(id result) {
    if ([result isKindOfClass[NSError error]]) {
        NSLog(@"Error: %@", error);
    }
    else {
        // execute on a certain execution context (main thread) if required:
        dispatch_async(dispatch_get_main_queue(), ^{
            APDUResponse* response = result;
            ...    
        });
    }
}];

警告:

不幸的是，使用运行循环通过底层任务正确地实现并发 NSOperation子类并不是那么简单。将会出现细微的并发问题，迫使您使用同步原语，例如锁或调度队列，以及一些其他技巧，以使其真正可靠。

幸运的是，将任何Run Loop任务包装到并发的 NSOperation子类中，基本上需要相同的“样板”代码。因此，一旦有了通用解决方案，编码工作就是从"template"进行复制/粘贴，然后为特定目的定制代码。

替代解决方案:

严格来说，如果您不打算将许多任务放入 NSOperation，则甚至不需要 NSOperationQueue的子类。并发操作可以简单地通过向其发送 start方法开始-不需要 NSOperationQueue。然后，不使用 NSOperation的子类可以使您自己的实现更简单，因为子类 NSOperation本身具有其自身的微妙之处。

但是，实际上，您需要一个“操作对象”来包装运行循环以驱动 NSStream对象，因为实现需要保持状态，而这是无法通过简单的异步方法完成的。

因此，您可以使用任何可被视为异步操作的自定义类，该类具有 start和 cancel方法，并具有在基础任务完成时通知调用站点的机制。

与完成处理程序相比，还有更强大的通知调用站点的方法。例如:promise或future(请参阅Wiki文章 Futures and promises)。

假设您使用Promise实现了自己的“异步操作”类，作为通知调用站点的一种方式，例如:

@interface WriteDataToStreamOperation : AsyncOperation

- (void) start;
- (void) cancel;

@property (nonatomic, readonly) BOOL isCancelled;
@property (nonatomic, readonly) BOOL isExecuting;
@property (nonatomic, readonly) BOOL isFinished;
@property (nonatomic, readonly) Promise* promise;

@end

您的原始问题看起来会更多“同步”-尽管仍然是异步的:

您的 sendCommand方法变为:

注意:假定Promise类的某个实现:

- (Promise*) sendCommand:(NSData *)command {
    WriteDataToStreamOperation* op = 
     [[WriteDataToStreamOperation alloc] initWithData:command 
                                         outputStream:self.outputStream];
    [op start];
    Promise* promise = op.promise;
    [promise setTimeout:100]; // time out after 100 seconds
    return promise;
}

注意: promise 已设置“超时”。这基本上是在注册计时器和处理程序。如果计时器在底层任务解决了 promise 之前触发了计时器，则计时器块将解决带有超时错误的 promise 。如何实现(以及是否实现)取决于Promise库。 (在这里，我假设我是作者的RXPromise库。其他实现也可以实现这种功能)。

用法:

[self sendCommand:request].then(^id(APDUResponse* response) {
    // do something with the response
    ...
    return  ...;  // returns the result of the handler
}, 
^id(NSError*error) {
    // A Stream error or a timeout error
    NSLog(@"Error: %@", error);
    return nil;  // returns nothing
});

替代用法:

您可以用其他方式设置超时时间。现在，假设我们没有在 sendCommand:方法中设置超时。

我们可以将超时设置为“外部”:

Promise* promise = [self sendCommand:request];
[promise setTimeout:100];
promise.then(^id(APDUResponse* response) {
    // do something with the response
    ...
    return  ...;  // returns the result of the handler
}, 
^id(NSError*error) {
    // A Stream error or a timeout error
    NSLog(@"Error: %@", error);
    return nil;  // returns nothing
});

使异步方法同步

通常，您不需要，也不应该在应用程序代码中将异步方法“转换”为某种同步方法。这总是导致次优且效率低下的代码，从而不必要地消耗了系统资源(例如线程)。

但是，您可能需要在有意义的单元测试中执行此操作:

单元测试中“同步”异步方法的示例

在测试实现时，您经常希望“等待”(同步是)以得到结果。您的基础任务实际上是在“运行循环”上执行的，实际上可能在您要等待结果的同一线程上执行，这一事实并没有使解决方案变得更简单。

但是，您可以使用 runLoopWait方法，通过RXPromise库轻松完成此操作，该方法有效地进入运行循环，并在该循环中等待可解决的 promise :

-(void) testSendingCommandShouldReturnResponseBeforeTimeout10 {
    Promise* promise = [self sendCommand:request];
    [promise setTimeout:10];
    [promise.then(^id(APDUResponse* response) {
        // do something with the response
        XCTAssertNotNil(response);            
        return  ...;  // returns the result of the handler
    }, 
    ^id(NSError*error) {
         // A Stream error or a timeout error
        XCTestFail(@"failed with error: %@", error);
        return nil;  // returns nothing

    }) runLoopWait];  // "wait" on the run loop
}

在这里，方法 runLoopWait将进入运行循环，并等待超时或由于基础任务解决了 promise 而导致的 promise 被解决。 Promise不会阻塞主线程，也不会轮询运行循环。解决 promise 后，它将仅离开运行循环。其他运行循环事件将照常处理。

注意:您可以从主线程安全地调用 testSendingCommandShouldReturnResponseBeforeTimeout10而不阻塞它。这是绝对必要的，因为您的Stream委托(delegate)方法也可以在主线程上执行!

单元测试库中通常还有其他方法，它们提供了与进入运行循环时“等待”异步方法或操作结果类似的功能。

不建议使用其他方法“等待”异步方法或操作的最终结果。这些通常会将方法分派(dispatch)给私有(private)线程，然后将其阻塞，直到结果可用为止。

有用的资源

类之类的操作的代码片段(在Gist上)，该片段使用Promises将流复制到另一个流中:
RXStreamToStreamCopier