natas_hw
/
grpc
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Polish GRXBufferedPipe

This commit is contained in:
Muxi Yan 2017-06-13 10:00:53 -07:00
parent d7d6a2e12b
commit d4792e9d8d
1 changed files with 9 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
src/objective-c/RxLibrary/GRXBufferedPipe.m
View File

@ -41,7 +41,7 @@
@implementation GRXBufferedPipe { @implementation GRXBufferedPipe {
NSError *_errorOrNil; NSError *_errorOrNil;
dispatch_queue_t _writeQueue; dispatch_queue_t _writeQueue;
dispatch_once_t _finishQueue; dispatch_once_t _finishAction;
} }
@synthesize state = _state; @synthesize state = _state;
@ -54,7 +54,6 @@
if (self = [super init]) { if (self = [super init]) {
_state = GRXWriterStateNotStarted; _state = GRXWriterStateNotStarted;
_writeQueue = dispatch_queue_create(NULL, DISPATCH_QUEUE_SERIAL); _writeQueue = dispatch_queue_create(NULL, DISPATCH_QUEUE_SERIAL);
self.inputIsFinished = NO;
dispatch_suspend(_writeQueue); dispatch_suspend(_writeQueue);
} }
return self; return self;
@ -74,10 +73,7 @@
} }
__weak GRXBufferedPipe *weakSelf = self; __weak GRXBufferedPipe *weakSelf = self;
dispatch_async(_writeQueue, ^(void) { dispatch_async(_writeQueue, ^(void) {
GRXBufferedPipe *strongSelf = weakSelf; [weakSelf.writeable writeValue:value];
if (strongSelf && strongSelf.writeable) {
[strongSelf.writeable writeValue:value];
}
}); });
} }
@ -86,23 +82,18 @@
return; return;
} }
self.inputIsFinished = YES; self.inputIsFinished = YES;
if (errorOrNil) { dispatch_once(&_finishAction, ^{
// No need to write pending values. if (errorOrNil) {
dispatch_once(&_finishQueue, ^{ // No need to write pending values.
_errorOrNil = errorOrNil; _errorOrNil = errorOrNil;
[self finishWithError:_errorOrNil]; [self finishWithError:_errorOrNil];
}); } else {
} else {
dispatch_once(&_finishQueue, ^{
__weak GRXBufferedPipe *weakSelf = self; __weak GRXBufferedPipe *weakSelf = self;
dispatch_async(_writeQueue, ^{ dispatch_async(_writeQueue, ^{
GRXBufferedPipe *strongSelf = weakSelf; [weakSelf finishWithError:nil];
if (strongSelf) {
[strongSelf finishWithError:nil];
}
}); });
}); }
} });
} }
#pragma mark GRXWriter implementation #pragma mark GRXWriter implementation