Refactor event listener to use async instead of Dispatcher

dom/src/main/scala/fs2/dom/EventTargetHelpers.scala

@@ -18,38 +18,28 @@ package fs2
 package dom
 
 import cats.effect.kernel.Async
-import cats.effect.kernel.Resource
-import cats.effect.std.Dispatcher
-import cats.syntax.all._
-import fs2.concurrent.Channel
-import org.scalajs.dom.Event
-import org.scalajs.dom.EventListenerOptions
-import org.scalajs.dom.EventTarget
+import org.scalajs.dom
 
 private[dom] object EventTargetHelpers {
 
-  def listen[F[_], E <: Event](target: EventTarget, `type`: String)(implicit
+  def listen[F[_], E <: dom.Event](target: dom.EventTarget, `type`: String)(implicit
       F: Async[F]
-  ): Resource[F, Stream[F, E]] = {
-    val setup = for {
-      dispatcher <- Dispatcher.sequential[F]
-      abort <- AbortController[F]
-      ch <- Resource.eval {
-        Channel.unbounded[F, E].flatTap { ch =>
-          F.delay {
-            target.addEventListener[E](
-              `type`,
-              (ev: E) => dispatcher.unsafeRunAndForget(ch.send(ev)),
-              new EventListenerOptions {
-                signal = abort
-              }
-            )
-          }
+  ): Stream[F, E] =
+    Stream.repeatEval {
+      F.async[E] { cb =>
+        F.delay {
+          val ctrl = new dom.AbortController
+          target.addEventListener[E](
+            `type`,
+            (e: E) => cb(Right(e)),
+            new dom.EventListenerOptions {
+              once = true
+              signal = ctrl.signal
+            }
+          )
+          Some(F.delay(ctrl.abort()))
         }
       }
-    } yield ch
-
-    setup.map(_.stream)
-  }
+    }
 
 }

dom/src/main/scala/fs2/dom/History.scala

@@ -19,6 +19,7 @@ package fs2.dom
 import cats.data.OptionT
 import cats.effect.kernel.Async
 import cats.effect.kernel.Ref
+import cats.effect.std.Queue
 import cats.syntax.all._
 import fs2.Stream
 import fs2.concurrent.Signal
@@ -51,10 +52,15 @@ object History {
     new History[F, S] {
 
       def state = new Signal[F, Option[S]] {
-        def discrete =
-          Stream.resource(eventsResource[F, PopStateEvent](window, "popstate")).flatMap { events =>
-            Stream.eval(get) ++ events.evalMap(e => serializer.deserialize(e.state).map(Some(_)))
-          }
+        def discrete = Stream.eval(Queue.circularBuffer[F, PopStateEvent](1)).flatMap { queue =>
+          val head = Stream.eval(get)
+          val tail =
+            Stream.repeatEval(queue.take).evalMap(e => serializer.deserialize(e.state).map(Some(_)))
+
+          val listener = events[F, PopStateEvent](window, "popstate").foreach(queue.offer(_))
+
+          (head ++ tail).concurrently(listener)
+        }
 
         def get = OptionT(F.delay(Option(window.history.state)))
           .semiflatMap(serializer.deserialize(_))

dom/src/main/scala/fs2/dom/package.scala

@@ -45,12 +45,12 @@ package object dom {
     stream.through(toReadableStream).compile.resource.lastOrError
 
   def events[F[_]: Async, E <: Event](target: EventTarget, `type`: String): Stream[F, E] =
-    Stream.resource(EventTargetHelpers.listen(target, `type`)).flatten
+    EventTargetHelpers.listen(target, `type`)
 
+  @deprecated("Use events", "0.1.1")
   def eventsResource[F[_]: Async, E <: Event](
       target: EventTarget,
       `type`: String
-  ): Resource[F, Stream[F, E]] =
-    EventTargetHelpers.listen(target, `type`)
+  ): Resource[F, Stream[F, E]] = Resource.pure(events(target, `type`))
 
 }

testsBrowser/src/test/scala/fs2/dom/HistorySuite.scala

@@ -39,6 +39,7 @@ class HistorySuite extends CatsEffectSuite {
           _ <- history.state.get.assertEquals(Some(3))
           _ <- history.back
           _ <- history.state.get.assertEquals(Some(1))
+          _ <- IO.sleep(1.second) // before the next UI
           _ <- history.forward
           _ <- history.state.get.assertEquals(Some(3))
           _ <- ch.stream.take(2).compile.toList.assertEquals(List(1, 3))