Merge pull request #117 from alexarchambault/merge-upstream

Merge upstream changes

Author: alexarchambault

backend/src/main/scala/bloop/BloopClassFileManager.scala

@@ -16,9 +16,9 @@ import bloop.io.AbsolutePath
 import bloop.io.ParallelOps
 import bloop.io.ParallelOps.CopyMode
 import bloop.io.{Paths => BloopPaths}
+import bloop.task.Task
 import bloop.tracing.BraveTracer
 
-import monix.eval.Task
 import xsbti.compile.ClassFileManager
 
 final class BloopClassFileManager(

backend/src/main/scala/bloop/CompileBackgroundTasks.scala

@@ -3,10 +3,9 @@ package bloop
 import bloop.io.AbsolutePath
 import bloop.logging.Logger
 import bloop.reporter.Reporter
+import bloop.task.Task
 import bloop.tracing.BraveTracer
 
-import monix.eval.Task
-
 abstract class CompileBackgroundTasks {
   def trigger(
       clientClassesDir: AbsolutePath,

backend/src/main/scala/bloop/Compiler.scala

@@ -19,12 +19,12 @@ import bloop.logging.ObservedLogger
 import bloop.reporter.ProblemPerPhase
 import bloop.reporter.Reporter
 import bloop.reporter.ZincReporter
+import bloop.task.Task
 import bloop.tracing.BraveTracer
 import bloop.util.AnalysisUtils
 import bloop.util.CacheHashCode
 import bloop.util.UUIDUtil
 
-import monix.eval.Task
 import monix.execution.Scheduler
 import sbt.internal.inc.Analysis
 import sbt.internal.inc.ConcreteAnalysisContents

backend/src/main/scala/bloop/io/ClasspathHasher.scala

@@ -13,11 +13,13 @@ import java.util.zip.ZipEntry
 
 import scala.collection.mutable
 import scala.concurrent.Promise
+import scala.util.control.NonFatal
 
 import bloop.logging.Logger
+import bloop.task.Task
 import bloop.tracing.BraveTracer
 
-import monix.eval.Task
+import monix.eval.{Task => MonixTask}
 import monix.execution.Cancelable
 import monix.execution.Scheduler
 import monix.execution.atomic.AtomicBoolean
@@ -76,10 +78,10 @@ object ClasspathHasher {
 
     val isCancelled = AtomicBoolean(false)
     val parallelConsumer = {
-      Consumer.foreachParallelAsync[AcquiredTask](parallelUnits) {
+      Consumer.foreachParallelTask[AcquiredTask](parallelUnits) {
         case AcquiredTask(path, idx, p) =>
           // Use task.now because Monix's load balancer already forces an async boundary
-          val hashingTask = Task.now {
+          val hashingTask = MonixTask.now {
             val hash =
               try {
                 if (cancelCompilation.isCompleted) {
@@ -108,7 +110,7 @@ object ClasspathHasher {
                 }
               } catch {
                 // Can happen when a file doesn't exist, for example
-                case monix.execution.misc.NonFatal(_) => BloopStamps.emptyHash(path)
+                case NonFatal(_) => BloopStamps.emptyHash(path)
               }
             classpathHashes(idx) = hash
             hashingPromises.remove(path, p)
@@ -147,8 +149,8 @@ object ClasspathHasher {
           )
 
           hashingTask
-            .doOnCancel(Task(timeoutCancellation.cancel()))
-            .doOnFinish(_ => Task(timeoutCancellation.cancel()))
+            .doOnCancel(MonixTask(timeoutCancellation.cancel()))
+            .doOnFinish(_ => MonixTask(timeoutCancellation.cancel()))
       }
     }
 
@@ -172,7 +174,7 @@ object ClasspathHasher {
                     // If the process that acquired it cancels the computation, try acquiring it again
                     logger
                       .warn(s"Unexpected hash computation of $entry was cancelled, restarting...")
-                    Task.fork(Task.eval(acquireHashingEntry(entry, entryIdx)))
+                    Task.eval(acquireHashingEntry(entry, entryIdx)).asyncBoundary
                   }
                 } else {
                   Task.now {

backend/src/main/scala/bloop/io/ParallelOps.scala

@@ -11,7 +11,9 @@ import java.util.concurrent.ConcurrentHashMap
 
 import scala.concurrent.Promise
 
-import monix.eval.Task
+import bloop.task.Task
+
+import monix.eval.{Task => MonixTask}
 import monix.execution.Cancelable
 import monix.execution.Scheduler
 import monix.execution.atomic.AtomicBoolean
@@ -148,7 +150,7 @@ object ParallelOps {
       Cancelable.cancelAll(completeSubscribers :: tasksToCancel)
     }
 
-    val copyFileSequentially = Consumer.foreachAsync[((Path, BasicFileAttributes), Path)] {
+    val copyFileSequentially = Consumer.foreachTask[((Path, BasicFileAttributes), Path)] {
       case ((originFile, originAttrs), targetFile) =>
         def copy(replaceExisting: Boolean): Unit = {
           if (replaceExisting) {
@@ -169,7 +171,7 @@ object ParallelOps {
         }
 
         // It's important that this task is not forked for performance reasons
-        def triggerCopy(p: Promise[Unit]) = Task.eval {
+        def triggerCopy(p: Promise[Unit]) = MonixTask.eval {
           try {
             // Skip work if cancellation is on and complete promise in finalizer
             if (isCancelled.get) ()
@@ -205,25 +207,27 @@ object ParallelOps {
           ()
         }
 
-        def acquireFile: Task[Unit] = {
+        def acquireFile: MonixTask[Unit] = {
           val currentPromise = Promise[Unit]()
           val promiseInMap = takenByOtherCopyProcess.putIfAbsent(originFile, currentPromise)
           if (promiseInMap == null) {
             triggerCopy(currentPromise)
           } else {
-            Task.fromFuture(promiseInMap.future).flatMap(_ => acquireFile)
+            MonixTask.fromFuture(promiseInMap.future).flatMap(_ => acquireFile)
           }
         }
 
         acquireFile.coeval(scheduler).value match {
           // The happy path is that we evaluate the task and return
-          case Right(()) => Task.now(())
+          case Right(()) => MonixTask.now(())
           case Left(cancelable) =>
             // Blocked on another process to finish the copy of a file, when it's done we restart
             cancelables.synchronized { cancelables.+=(cancelable) }
-            Task
+            MonixTask
               .fromFuture(cancelable)
-              .doOnFinish(_ => Task { cancelables.synchronized { cancelables.-=(cancelable) }; () })
+              .doOnFinish(_ =>
+                MonixTask { cancelables.synchronized { cancelables.-=(cancelable) }; () }
+              )
         }
     }