Remove Use Of Interruptible

build.sbt

@@ -36,7 +36,7 @@ Global / onChangedBuildSource := ReloadOnSourceChanges
 addCommandAlias("fmt", "all scalafmtSbt scalafmt test:scalafmt")
 addCommandAlias("check", "all scalafmtSbtCheck scalafmtCheck test:scalafmtCheck")
 
-val zioVersion        = "2.0.0"
+val zioVersion        = "2.0.2"
 val rsVersion         = "1.0.4"
 val collCompatVersion = "2.7.0"
 

src/main/scala/zio/interop/reactivestreams/Adapters.scala

@@ -21,7 +21,7 @@ object Adapters {
       if (subscriber == null) {
         throw new NullPointerException("Subscriber must not be null.")
       } else
-        unsafeCompat { implicit unsafe =>
+        unsafe { implicit unsafe =>
           val subscription = new DemandTrackingSubscription(subscriber)
           runtime.unsafe.fork(
             for {
@@ -39,7 +39,7 @@ object Adapters {
   def subscriberToSink[E <: Throwable, I](
     subscriber: => Subscriber[I]
   )(implicit trace: Trace): ZIO[Scope, Nothing, (E => UIO[Unit], ZSink[Any, Nothing, I, I, Unit])] =
-    unsafeCompat { implicit unsafe =>
+    unsafe { implicit unsafe =>
       val sub = subscriber
       for {
         error       <- Promise.make[E, Nothing]
@@ -59,7 +59,7 @@ object Adapters {
         subscriberP    <- makeSubscriber[O](bufferSize)
         (subscriber, p) = subscriberP
         _              <- ZIO.acquireRelease(ZIO.succeed(publisher.subscribe(subscriber)))(_ => ZIO.succeed(subscriber.interrupt()))
-        subQ           <- p.await.interruptible
+        subQ           <- p.await
         (sub, q)        = subQ
         process        <- process(sub, q, () => subscriber.await(), () => subscriber.isDone)
       } yield process
@@ -134,7 +134,7 @@ object Adapters {
            )(
              _.poll.flatMap(_.fold(ZIO.unit)(_.foldZIO(_ => ZIO.unit, { case (sub, _) => ZIO.succeed(sub.cancel()) })))
            )
-    } yield unsafeCompat { implicit unsafe =>
+    } yield unsafe { implicit unsafe =>
       val subscriber =
         new InterruptibleSubscriber[A] {
 

src/test/scala/zio/interop/reactivestreams/PublisherToStreamSpec.scala

@@ -96,7 +96,7 @@ object PublisherToStreamSpec extends ZIOSpecDefault {
       } @@ TestAspect.timeout(3.seconds),
       test("cancels subscription when interrupted before subscription") {
         val tst =
-          Unsafe.unsafeCompat { implicit unsafe =>
+          Unsafe.unsafe { implicit unsafe =>
             for {
               subscriberP    <- Promise.make[Nothing, Subscriber[_]]
               cancelledLatch <- Promise.make[Nothing, Unit]

src/test/scala/zio/interop/reactivestreams/SinkToSubscriberSpec.scala

@@ -85,7 +85,7 @@ object SinkToSubscriberSpec extends ZIOSpecDefault {
     )
 
   val makePublisherProbe =
-    Unsafe.unsafeCompat { implicit unsafe =>
+    Unsafe.unsafe { implicit unsafe =>
       for {
         subscribed <- Promise.make[Nothing, Unit]
         requested  <- Promise.make[Nothing, Unit]

src/test/scala/zio/interop/reactivestreams/StreamToPublisherSpec.scala

@@ -21,7 +21,7 @@ object StreamToPublisherSpec extends ZIOSpecDefault {
     new PublisherVerification[Int](new TestEnvironment(2000, 500), 2000L) {
 
       def createPublisher(elements: Long): Publisher[Int] =
-        Unsafe.unsafeCompat { implicit unsafe =>
+        Unsafe.unsafe { implicit unsafe =>
           runtime.unsafe
             .run(
               ZStream
@@ -32,7 +32,7 @@ object StreamToPublisherSpec extends ZIOSpecDefault {
         }
 
       override def createFailedPublisher(): Publisher[Int] =
-        Unsafe.unsafeCompat { implicit unsafe =>
+        Unsafe.unsafe { implicit unsafe =>
           runtime.unsafe
             .run(
               ZStream