 	}
 
 	/// Initializes a pipe object using existing file descriptors.
-	init(readFD: Int32, writeFD: Int32, queue: dispatch_queue_t) {
+	init(readFD: Int32, writeFD: Int32, queue: dispatch_queue_t, group: dispatch_group_t) {
 		precondition(readFD >= 0)
 		precondition(writeFD >= 0)
 
 		self.readFD = readFD
 		self.writeFD = writeFD
 		self.queue = queue
+		self.group = group
 	}
 
 	/// Instantiates a new descriptor pair.
-	class func create(queue: dispatch_queue_t) -> Result<Pipe, ReactiveTaskError> {
+	class func create(queue: dispatch_queue_t, _ group: dispatch_group_t) -> Result<Pipe, ReactiveTaskError> {
 		var fildes: [Int32] = [ 0, 0 ]
 		if pipe(&fildes) == 0 {
-			return .success(self(readFD: fildes[0], writeFD: fildes[1], queue: queue))
+			return .success(self(readFD: fildes[0], writeFD: fildes[1], queue: queue, group: group))
 		} else {
 			return .failure(.POSIXError(errno))
 		}

										}
										|> start(observer)
								}
							}

							task.launch()

				}

				close(self.writeFD)
				dispatch_group_leave(self.group)
			}

			producer.startWithSignal { signal, producerDisposable in
