Merge pull request #657 from adpi2/hide-reactivex

Hide reactivex from api

Author: adpi2

modules/core/src/main/scala/ch/epfl/scala/debugadapter/Debuggee.scala

@@ -2,7 +2,7 @@ package ch.epfl.scala.debugadapter
 
 import java.nio.file.Path
 import java.io.File
-import io.reactivex.Observable
+import java.io.Closeable
 
 trait Debuggee {
   def name: String
@@ -12,7 +12,7 @@ trait Debuggee {
   def libraries: Seq[Library]
   def unmanagedEntries: Seq[UnmanagedEntry]
   def javaRuntime: Option[JavaRuntime]
-  def classesToUpdate: Observable[Seq[String]]
+  def observeClassesToUpdate(updateClasses: Seq[String] => Unit): Closeable
 
   def managedEntries: Seq[ManagedEntry] = modules ++ libraries
   def classPathEntries: Seq[ClassPathEntry] = managedEntries ++ unmanagedEntries

modules/core/src/main/scala/ch/epfl/scala/debugadapter/internal/HotCodeReplaceProvider.scala

@@ -15,7 +15,6 @@ import com.microsoft.java.debug.core.adapter.IStackTraceProvider
 import com.microsoft.java.debug.core.protocol.Events
 import com.sun.jdi.*
 import io.reactivex.Observable
-import io.reactivex.disposables.Disposable
 
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.atomic.AtomicReference
@@ -27,6 +26,7 @@ import scala.util.Success
 import scala.util.Try
 import scala.util.control.NonFatal
 import io.reactivex.subjects.PublishSubject
+import java.io.Closeable
 
 class HotCodeReplaceProvider(
     debuggee: Debuggee,
@@ -35,7 +35,7 @@ class HotCodeReplaceProvider(
 ) extends IHotCodeReplaceProvider {
   private var sourceLookUp: SourceLookUpProvider = null
   private var stackTraceProvider: StackTraceProvider = null
-  private var subscription: Disposable = null
+  private var subscription: Closeable = null
   private var context: IDebugAdapterContext = null
   private var currentDebugSession: IDebugSession = null
   private val classesAccumulator: AtomicReference[Set[String]] = new AtomicReference(Set.empty)
@@ -45,14 +45,13 @@ class HotCodeReplaceProvider(
   override def initialize(context: IDebugAdapterContext, options: ju.Map[String, Object]): Unit = {
     this.context = context
     this.currentDebugSession = context.getDebugSession
-    this.subscription = debuggee.classesToUpdate.subscribe(classes => classesAccumulator.updateAndGet(_ ++ classes))
+    this.subscription = debuggee.observeClassesToUpdate(classes => classesAccumulator.updateAndGet(_ ++ classes))
     this.sourceLookUp = context.getProvider(classOf[ISourceLookUpProvider]).asInstanceOf[SourceLookUpProvider]
     this.stackTraceProvider = context.getProvider(classOf[IStackTraceProvider]).asInstanceOf[StackTraceProvider]
-
   }
 
   override def close(): Unit = {
-    if (subscription != null) subscription.dispose()
+    if (subscription != null) subscription.close()
     context = null
     currentDebugSession = null
     subscription = null

modules/sbt-plugin/src/main/scala/ch/epfl/scala/debugadapter/sbtplugin/DebugAdapterPlugin.scala

@@ -20,7 +20,6 @@ import sjsonnew.support.scalajson.unsafe.CompactPrinter
 import sjsonnew.support.scalajson.unsafe.Converter
 import sjsonnew.support.scalajson.unsafe.{Parser => JsonParser}
 import xsbti.FileConverter
-import xsbti.VirtualFileRef
 import xsbti.compile.CompileAnalysis
 import xsbti.compile.analysis.ReadStamps
 import xsbti.compile.analysis.Stamp
@@ -67,7 +66,7 @@ object DebugAdapterPlugin extends sbt.AutoPlugin {
       inputKey[URI]("Start a debug session on a remote process").withRank(KeyRanks.DTask)
     val debugAdapterConfig =
       settingKey[DebugConfig]("Configure the debug session").withRank(KeyRanks.DTask)
-    val debugAdapterClassesObserver =
+    val debugAdapterClassesToUpdate =
       settingKey[PublishSubject[Seq[String]]]("Observe the classes to be reloaded by the debuggee")
         .withRank(KeyRanks.DTask)
     val stopDebugSession =
@@ -109,7 +108,7 @@ object DebugAdapterPlugin extends sbt.AutoPlugin {
     startMainClassDebugSession := mainClassSessionTask.evaluated,
     startRemoteDebugSession := remoteSessionTask.evaluated,
     stopDebugSession := stopSessionTask.value,
-    debugAdapterClassesObserver := PublishSubject.create(),
+    debugAdapterClassesToUpdate := PublishSubject.create(),
     Keys.compile / Keys.javacOptions := {
       val jo = (Keys.compile / Keys.javacOptions).value
       if (jo.exists(_.startsWith("-g"))) jo
@@ -118,45 +117,38 @@ object DebugAdapterPlugin extends sbt.AutoPlugin {
     Keys.compile := {
       val currentAnalysis: CompileAnalysis = Keys.compile.value
       val previousAnalysis = Keys.previousCompile.value.analysis
-      val observer = debugAdapterClassesObserver.value
+      val classesToUpdate = debugAdapterClassesToUpdate.value
       val fileConverter = Keys.fileConverter.value
       val classDir = Keys.classDirectory.value.toPath
       if (previousAnalysis.isPresent) {
-        val classesToReload =
-          getNewClasses(currentAnalysis.readStamps, previousAnalysis.get.readStamps, fileConverter, classDir)
-        observer.onNext(classesToReload)
+        val currentStamps = currentAnalysis.readStamps
+        val previousStamps = previousAnalysis.get.readStamps
+        val newClasses = getNewClasses(currentStamps, previousStamps, fileConverter, classDir)
+        classesToUpdate.onNext(newClasses)
       }
       currentAnalysis
     }
   )
 
-  def getNewClasses(
+  private def getNewClasses(
       currentStamps: ReadStamps,
       previousStamps: ReadStamps,
-      fileConverter: FileConverter,
+      converter: FileConverter,
       classDir: java.nio.file.Path
   ): Seq[String] = {
-    def isNewer(current: Stamp, previous: Stamp) = {
+    def isNewer(current: Stamp, previous: Stamp) =
       if (previous == null) true
       else {
         val newHash = current.getHash
         val oldHash = previous.getHash
         newHash.isPresent && (!oldHash.isPresent || newHash.get != oldHash.get)
       }
-    }
-
-    object ClassFile {
-      def unapply(vf: VirtualFileRef): Option[String] = {
-        val path = fileConverter.toPath(vf)
-        if (path.toString.endsWith(".class")) {
-          Some(classDir.relativize(path).toString.replace(File.separator, ".").stripSuffix(".class"))
-        } else None
-      }
-    }
 
     val oldStamps = previousStamps.getAllProductStamps
-    currentStamps.getAllProductStamps.asScala.collect {
-      case (file @ ClassFile(fqcn), stamp) if isNewer(stamp, oldStamps.get(file)) => fqcn
+    currentStamps.getAllProductStamps.asScala.iterator.collect {
+      case (file, stamp) if file.id.endsWith(".class") && isNewer(stamp, oldStamps.get(file)) =>
+        val path = converter.toPath(file)
+        classDir.relativize(path).toString.replace(File.separator, ".").stripSuffix(".class")
     }.toSeq
   }
 
@@ -291,7 +283,7 @@ object DebugAdapterPlugin extends sbt.AutoPlugin {
             InternalTasks.libraries.value,
             InternalTasks.unmanagedEntries.value,
             InternalTasks.javaRuntime.value,
-            InternalTasks.classesObservable.value,
+            InternalTasks.classesToUpdate.value,
             mainClass.`class`,
             mainClass.arguments,
             new LoggerAdapter(logger)
@@ -397,7 +389,7 @@ object DebugAdapterPlugin extends sbt.AutoPlugin {
             InternalTasks.libraries.value,
             InternalTasks.unmanagedEntries.value,
             InternalTasks.javaRuntime.value,
-            InternalTasks.classesObservable.value,
+            InternalTasks.classesToUpdate.value,
             cleanups,
             parallel,
             testRunners,
@@ -463,7 +455,7 @@ object DebugAdapterPlugin extends sbt.AutoPlugin {
           InternalTasks.libraries.value,
           InternalTasks.unmanagedEntries.value,
           InternalTasks.javaRuntime.value,
-          InternalTasks.classesObservable.value,
+          InternalTasks.classesToUpdate.value,
           new LoggerAdapter(logger)
         )
       startServer(jobService, scope, state, target, debuggee, debugToolsResolver, debugAdapterConfig.value)

modules/sbt-plugin/src/main/scala/ch/epfl/scala/debugadapter/sbtplugin/internal/InternalTasks.scala

@@ -17,12 +17,12 @@ private[sbtplugin] object InternalTasks {
     modules.join(_.join)
   }
 
-  lazy val classesObservable: Def.Initialize[Observable[Seq[String]]] = Def.settingDyn {
+  lazy val classesToUpdate: Def.Initialize[Observable[Seq[String]]] = Def.settingDyn {
     val internalDependencies = Keys.bspInternalDependencyConfigurations
     val observables = for {
       (proj, configs) <- Keys.bspInternalDependencyConfigurations.value
       config <- configs
-    } yield (proj / config / debugAdapterClassesObserver)
+    } yield (proj / config / debugAdapterClassesToUpdate)
     Def.setting {
       observables.join.value.fold(Observable.empty[Seq[String]])(_ mergeWith _)
     }

modules/sbt-plugin/src/main/scala/ch/epfl/scala/debugadapter/sbtplugin/internal/SbtDebuggee.scala

@@ -14,9 +14,16 @@ import scala.concurrent.{ExecutionContext, Future}
 import scala.util.control.NonFatal
 import ch.epfl.scala.debugadapter.testing.TestSuiteEvent
 import io.reactivex.Observable
+import java.io.Closeable
 
 private[debugadapter] sealed trait SbtDebuggee extends Debuggee {
   val logger: LoggerAdapter
+  val classesToUpdate: Observable[Seq[String]]
+
+  override def observeClassesToUpdate(updateClasses: Seq[String] => Unit): Closeable = {
+    val subscription = classesToUpdate.subscribe(updateClasses(_))
+    () => if (!subscription.isDisposed) subscription.dispose
+  }
 }
 
 private[debugadapter] final class MainClassDebuggee(

modules/tests/src/main/scala/ch/epfl/scala/debugadapter/testfmk/MockDebuggee.scala

@@ -11,8 +11,7 @@ import ch.epfl.scala.debugadapter.Module
 import ch.epfl.scala.debugadapter.Library
 import ch.epfl.scala.debugadapter.UnmanagedEntry
 import ch.epfl.scala.debugadapter.JavaRuntime
-import io.reactivex.subjects.PublishSubject
-import io.reactivex.Observable
+import java.io.Closeable
 
 class MockDebuggee extends Debuggee {
   val scalaVersion: ScalaVersion = ScalaVersion.`2.12`
@@ -35,7 +34,7 @@ class MockDebuggee extends Debuggee {
 
   override def javaRuntime: Option[JavaRuntime] = None
 
-  override val classesToUpdate: Observable[Seq[String]] = PublishSubject.create[Seq[String]]()
+  override def observeClassesToUpdate(updateClasses: Seq[String] => Unit): Closeable = () => ()
 }
 
 class MockCancelableFuture() extends CancelableFuture[Unit] {

modules/tests/src/main/scala/ch/epfl/scala/debugadapter/testfmk/TestingDebuggee.scala

@@ -12,8 +12,11 @@ import ch.epfl.scala.debugadapter.SourceDirectory
 import ch.epfl.scala.debugadapter.StandaloneSourceFile
 import ch.epfl.scala.debugadapter.UnmanagedEntry
 import ch.epfl.scala.debugadapter.testfmk.TestingDebuggee._
+import io.reactivex.subjects.PublishSubject
+import io.reactivex.subjects.Subject
 
 import java.io.BufferedReader
+import java.io.Closeable
 import java.io.InputStream
 import java.io.InputStreamReader
 import java.net.InetSocketAddress
@@ -25,8 +28,6 @@ import scala.concurrent.Future
 import scala.concurrent.Promise
 import scala.util.Properties
 import scala.util.control.NonFatal
-import io.reactivex.subjects.Subject
-import io.reactivex.subjects.PublishSubject
 
 case class TestingDebuggee(
     tempDir: Path,
@@ -41,7 +42,11 @@ case class TestingDebuggee(
 ) extends Debuggee
     with TestingContext {
 
-  override val classesToUpdate: Subject[Seq[String]] = PublishSubject.create()
+  val classesToUpdate: Subject[Seq[String]] = PublishSubject.create()
+  override def observeClassesToUpdate(updateClasses: Seq[String] => Unit): Closeable = {
+    val subscription = classesToUpdate.subscribe(updateClasses(_))
+    () => if (!subscription.isDisposed) subscription.dispose
+  }
   def mainSource: Path = sourceFiles.head
 
   override def name: String = mainClass