Some fixes for tests and change contract

2 files changed, 6 insertions, 6 deletions

diff --git a/src/objective-c/RxLibrary/GRXBufferedPipe.h b/src/objective-c/RxLibrary/GRXBufferedPipe.h
index bd7d4ad691..a871ea895a 100644
--- a/src/objective-c/RxLibrary/GRXBufferedPipe.h
+++ b/src/objective-c/RxLibrary/GRXBufferedPipe.h
@@ -27,8 +27,8 @@
  * immediately, unless flow control prevents it.
  * If it is throttled and keeps receiving values, as well as if it receives values before being
  * started, it will buffer them and propagate them in order as soon as its state becomes Started.
- * If it receives an error (via -writesFinishedWithError:), it will drop any buffered values and
- * propagate the error immediately.
+ * If it receives an end of stream (via -writesFinishedWithError:), it will buffer the EOS after the
+ * last buffered value and issue it to the writeable after all buffered values are issued.
  *
  * Beware that a pipe of this type can't prevent receiving more values when it is paused (for
  * example if used to write data to a congested network connection). Because in such situations the
diff --git a/src/objective-c/RxLibrary/GRXBufferedPipe.m b/src/objective-c/RxLibrary/GRXBufferedPipe.m
index 6433b62578..99cb0ad971 100644
--- a/src/objective-c/RxLibrary/GRXBufferedPipe.m
+++ b/src/objective-c/RxLibrary/GRXBufferedPipe.m
@@ -60,7 +60,7 @@
 - (void)writesFinishedWithError:(NSError *)errorOrNil {
   __weak GRXBufferedPipe *weakSelf = self;
   dispatch_async(_writeQueue, ^{
-    [weakSelf finishWithError:nil];
+    [weakSelf finishWithError:errorOrNil];
   });
 }
 
@@ -88,8 +88,7 @@
         }
         return;
       case GRXWriterStateStarted:
-        if (_state == GRXWriterStatePaused ||
-            _state == GRXWriterStateNotStarted) {
+        if (_state == GRXWriterStatePaused) {
           _state = newState;
           dispatch_resume(_writeQueue);
         }
@@ -102,7 +101,8 @@
 
 - (void)startWithWriteable:(id<GRXWriteable>)writeable {
   self.writeable = writeable;
-  self.state = GRXWriterStateStarted;
+  _state = GRXWriterStateStarted;
+  dispatch_resume(_writeQueue);
 }
 
 - (void)finishWithError:(NSError *)errorOrNil {