From 860b1da060e2250aaff905d3af7647b017f7a1ad Mon Sep 17 00:00:00 2001
From: Muxi Yan <mxyan@google.com>
Date: Sat, 29 Jul 2017 12:05:19 -0700
Subject: [PATCH] Resume GRXBufferedPipe when it gets dealloced

---
 src/objective-c/RxLibrary/GRXBufferedPipe.m |  8 +++
 src/objective-c/tests/RxLibraryUnitTests.m  | 70 +++++++++++++++++++++
 2 files changed, 78 insertions(+)

diff --git a/src/objective-c/RxLibrary/GRXBufferedPipe.m b/src/objective-c/RxLibrary/GRXBufferedPipe.m
index 99cb0ad9713..577a5e9a42b 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 fa3ded4c0cd..3a5adbbf378 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