Resume GRXBufferedPipe when it gets dealloced

2 files changed, 78 insertions, 0 deletions

diff --git a/src/objective-c/RxLibrary/GRXBufferedPipe.m b/src/objective-c/RxLibrary/GRXBufferedPipe.m
index 99cb0ad971..577a5e9a42 100644
--- a/src/objective-c/RxLibrary/GRXBufferedPipe.m
+++ b/src/objective-c/RxLibrary/GRXBufferedPipe.m
@@ -110,4 +110,12 @@
   self.state = GRXWriterStateFinished;
 }
 
+- (void)dealloc {
+  GRXWriterState state = self.state;
+  if (state == GRXWriterStateNotStarted ||
+      state == GRXWriterStatePaused) {
+    dispatch_resume(_writeQueue);
+  }
+}
+
 @end
diff --git a/src/objective-c/tests/RxLibraryUnitTests.m b/src/objective-c/tests/RxLibraryUnitTests.m
index fa3ded4c0c..3a5adbbf37 100644
--- a/src/objective-c/tests/RxLibraryUnitTests.m
+++ b/src/objective-c/tests/RxLibraryUnitTests.m
@@ -213,4 +213,74 @@
   XCTAssertEqualObjects(handler.errorOrNil, nil);
 }
 
+#define WRITE_ROUNDS (1000)
+- (void)testBufferedPipeResumeWhenDealloc {
+  id anyValue = @7;
+  id<GRXWriteable> writeable = [GRXWriteable  writeableWithSingleHandler:^(id value, NSError *errorOrNil) {
+  }];
+
+  // Release after alloc;
+  GRXBufferedPipe *pipe = [GRXBufferedPipe pipe];
+  pipe = nil;
+
+  // Release after write but before start
+  pipe = [GRXBufferedPipe pipe];
+  for (int i = 0; i < WRITE_ROUNDS; i++) {
+    [pipe writeValue:anyValue];
+  }
+  pipe = nil;
+
+  // Release after start but not write
+  pipe = [GRXBufferedPipe pipe];
+  [pipe startWithWriteable:writeable];
+  pipe = nil;
+
+  // Release after start and write
+  pipe = [GRXBufferedPipe pipe];
+  for (int i = 0; i < WRITE_ROUNDS; i++) {
+    [pipe writeValue:anyValue];
+  }
+  [pipe startWithWriteable:writeable];
+  pipe = nil;
+
+  // Release after start, write and pause
+  pipe = [GRXBufferedPipe pipe];
+  [pipe startWithWriteable:writeable];
+  for (int i = 0; i < WRITE_ROUNDS; i++) {
+    [pipe writeValue:anyValue];
+  }
+  pipe.state = GRXWriterStatePaused;
+  for (int i = 0; i < WRITE_ROUNDS; i++) {
+    [pipe writeValue:anyValue];
+  }
+  pipe = nil;
+
+  // Release after start, write, pause and finish
+  pipe = [GRXBufferedPipe pipe];
+  [pipe startWithWriteable:writeable];
+  for (int i = 0; i < WRITE_ROUNDS; i++) {
+    [pipe writeValue:anyValue];
+  }
+  pipe.state = GRXWriterStatePaused;
+  for (int i = 0; i < WRITE_ROUNDS; i++) {
+    [pipe writeValue:anyValue];
+  }
+  [pipe finishWithError:nil];
+  pipe = nil;
+
+  // Release after start, write, pause, finish and resume
+  pipe = [GRXBufferedPipe pipe];
+  [pipe startWithWriteable:writeable];
+  for (int i = 0; i < WRITE_ROUNDS; i++) {
+    [pipe writeValue:anyValue];
+  }
+  pipe.state = GRXWriterStatePaused;
+  for (int i = 0; i < WRITE_ROUNDS; i++) {
+    [pipe writeValue:anyValue];
+  }
+  [pipe finishWithError:nil];
+  pipe.state = GRXWriterStateStarted;
+  pipe = nil;
+}
+
 @end