Fix memory leak in continuous topic.

Partially reverts yet another few changes from typelevel@e25a6e1
doytsujin · May 19, 2021 · 41d2afe · 41d2afe
1 parent 989f7a2
commit 41d2afe
Showing 1 changed file with 4 additions and 2 deletions.
diff --git a/core/shared/src/main/scala/fs2/Pull.scala b/core/shared/src/main/scala/fs2/Pull.scala
@@ -1063,16 +1063,18 @@ object Pull extends PullLowPriority {
           else
             F.pure(extendedTopLevelScope)
 
+        val runner = new BuildR[G, X, End]
         val vrun = new ViewRunner(view)
         val tail = maybeCloseExtendedScope.flatMap { newExtendedScope =>
           scope.open(useInterruption).rethrow.flatMap { childScope =>
             val bb = new Bind[G, X, Unit, Unit](stream) {
               def cont(r: Terminal[Unit]): Pull[G, X, Unit] = endScope(childScope.id, r)
             }
-            go(childScope, newExtendedScope, translation, vrun, bb)
+            go(childScope, newExtendedScope, translation, runner, bb)
           }
         }
-        interruptGuard(scope, view, vrun)(tail)
+        interruptGuard(scope, view, runner)(tail).attempt
+          .flatMap(_.fold(goErr(_, view), _.apply(vrun)))
       }
 
       def goCloseScope(close: CloseScope, view: Cont[Unit, G, X]): F[End] = {