Polish comments and correct concurrent writeable behavior

5 files changed, 24 insertions, 18 deletions

diff --git a/src/objective-c/RxLibrary/GRXBufferedPipe.h b/src/objective-c/RxLibrary/GRXBufferedPipe.h
index a871ea895a..ae08cc315b 100644
--- a/src/objective-c/RxLibrary/GRXBufferedPipe.h
+++ b/src/objective-c/RxLibrary/GRXBufferedPipe.h
@@ -36,8 +36,7 @@
  * crash. If you want to react to flow control signals to prevent that, instead of using this class
  * you can implement an object that conforms to GRXWriter.
  *
- * Thread-safety:
- * The methods of an object of this class should not be called concurrently from different threads.
+ * Thread-safety: the methods of this class are thread-safe.
  */
 @interface GRXBufferedPipe : GRXWriter<GRXWriteable>
 
diff --git a/src/objective-c/RxLibrary/GRXConcurrentWriteable.h b/src/objective-c/RxLibrary/GRXConcurrentWriteable.h
index 606f5dea95..55468b3c07 100644
--- a/src/objective-c/RxLibrary/GRXConcurrentWriteable.h
+++ b/src/objective-c/RxLibrary/GRXConcurrentWriteable.h
@@ -43,21 +43,22 @@
 - (instancetype)initWithWriteable:(id<GRXWriteable>)writeable;
 
 /**
- * Enqueues writeValue: to be sent to the writeable in the main thread.
- * The passed handler is invoked from the main thread after writeValue: returns.
+ * Enqueues writeValue: to be sent to the writeable from the designated dispatch queue.
+ * The passed handler is invoked from designated dispatch queue after writeValue: returns.
  */
 - (void)enqueueValue:(id)value completionHandler:(void (^)(void))handler;
 
 /**
- * Enqueues writesFinishedWithError:nil to be sent to the writeable in the main thread. After that
- * message is sent to the writeable, all other methods of this object are effectively noops.
+ * Enqueues writesFinishedWithError:nil to be sent to the writeable in the designated dispatch
+ * queue. After that message is sent to the writeable, all other methods of this object are
+ * effectively noops.
  */
 - (void)enqueueSuccessfulCompletion;
 
 /**
  * If the writeable has not yet received a writesFinishedWithError: message, this will enqueue one
- * to be sent to it in the main thread, and cancel all other pending messages to the writeable
- * enqueued by this object (both past and future).
+ * to be sent to it in the designated dispatch queue, and cancel all other pending messages to the
+ * writeable enqueued by this object (both past and future).
  * The error argument cannot be nil.
  */
 - (void)cancelWithError:(NSError *)error;
diff --git a/src/objective-c/RxLibrary/GRXConcurrentWriteable.m b/src/objective-c/RxLibrary/GRXConcurrentWriteable.m
index 229d592f48..d9d0e8c31e 100644
--- a/src/objective-c/RxLibrary/GRXConcurrentWriteable.m
+++ b/src/objective-c/RxLibrary/GRXConcurrentWriteable.m
@@ -63,8 +63,10 @@
 
 - (void)enqueueSuccessfulCompletion {
   dispatch_async(_writeableQueue, ^{
-    if (self->_alreadyFinished) {
-      return;
+    @synchronized (self) {
+      if (self->_alreadyFinished) {
+        return;
+      }
     }
     [self.writeable writesFinishedWithError:nil];
     // Skip any possible message to the wrapped writeable enqueued after this one.
@@ -74,10 +76,14 @@
 
 - (void)cancelWithError:(NSError *)error {
   NSAssert(error != nil, @"For a successful completion, use enqueueSuccessfulCompletion.");
-  dispatch_async(_writeableQueue, ^{
+  @synchronized (self) {
     if (self->_alreadyFinished) {
       return;
     }
+  }
+  dispatch_async(_writeableQueue, ^{
+    // If enqueueSuccessfulCompletion is already issued, self.writeable is nil and the following
+    // line is no-op.
     [self.writeable writesFinishedWithError:error];
     self.writeable = nil;
   });
@@ -85,8 +91,10 @@
 
 - (void)cancelSilently {
   dispatch_async(_writeableQueue, ^{
-    if (self->_alreadyFinished) {
-      return;
+    @synchronized (self) {
+      if (self->_alreadyFinished) {
+        return;
+      }
     }
     self.writeable = nil;
   });
diff --git a/src/objective-c/RxLibrary/GRXForwardingWriter.h b/src/objective-c/RxLibrary/GRXForwardingWriter.h
index 3814ff8831..00366b6416 100644
--- a/src/objective-c/RxLibrary/GRXForwardingWriter.h
+++ b/src/objective-c/RxLibrary/GRXForwardingWriter.h
@@ -25,11 +25,7 @@
  * input writer, and for classes that represent objects with input and
  * output sequences of values, like an RPC.
  *
- * Thread-safety:
- * All messages sent to this object need to be serialized. When it is started, the writer it wraps
- * is started in the same thread. Manual state changes are propagated to the wrapped writer in the
- * same thread too. Importantly, all messages the wrapped writer sends to its writeable need to be
- * serialized with any message sent to this object.
+ * Thread-safety: the methods of this class are thread safe.
  */
 @interface GRXForwardingWriter : GRXWriter
 - (instancetype)initWithWriter:(GRXWriter *)writer NS_DESIGNATED_INITIALIZER;
diff --git a/src/objective-c/RxLibrary/GRXImmediateSingleWriter.h b/src/objective-c/RxLibrary/GRXImmediateSingleWriter.h
index 601abdc6b9..2fa38b3dce 100644
--- a/src/objective-c/RxLibrary/GRXImmediateSingleWriter.h
+++ b/src/objective-c/RxLibrary/GRXImmediateSingleWriter.h
@@ -23,6 +23,8 @@
 /**
  * Utility to construct GRXWriter instances from values that are immediately available when
  * required.
+ *
+ * Thread safety: the methods of this class are thread safe.
  */
 @interface GRXImmediateSingleWriter : GRXImmediateWriter