Add blockTimer and rename to blockShift

README.md

@@ -34,7 +34,7 @@ object ThreadNameExample {
     .map(Linebacker.fromExecutorService[IO](_)) // Create Linebacker From Executor
     .flatMap { implicit linebacker => // Raise Implicitly
       Stream.eval(
-        Linebacker[IO].block(getThread) // Block On Linebacker Pool Not Global
+        Linebacker[IO].blockShift(getThread) // Block On Linebacker Pool Not Global
       ) ++
       Stream.eval(getThread) // Running On Global
     }

build.sbt

@@ -28,13 +28,12 @@ lazy val commonSettings = Seq(
   organization := "io.chrisdavenport",
   scalaVersion := "2.12.6",
   crossScalaVersions := Seq(scalaVersion.value, "2.11.12"),
-  addCompilerPlugin("org.spire-math" % "kind-projector" % "0.9.6" cross CrossVersion.binary),
+  addCompilerPlugin("org.spire-math" % "kind-projector" % "0.9.7" cross CrossVersion.binary),
   scalafmtOnCompile := true,
   scalafmtTestOnCompile := true,
   libraryDependencies ++= Seq(
-    "org.typelevel" %% "cats-effect" % "0.10.1",
-    "co.fs2"        %% "fs2-core"    % "0.10.3",
-    "org.specs2"    %% "specs2-core" % "4.0.3" % Test
+    "org.typelevel" %% "cats-effect" % "1.0.0-RC2-d7181dc",
+    "org.specs2"    %% "specs2-core" % "4.2.0" % Test
   )
 )
 

linebacker/src/main/scala/io/chrisdavenport/linebacker/DualContext.scala

@@ -1,7 +1,6 @@
 package io.chrisdavenport.linebacker
 
 import cats._
-import cats.implicits._
 import cats.effect._
 import java.util.concurrent.ExecutorService
 import scala.concurrent.ExecutionContext
@@ -11,12 +10,7 @@ trait DualContext[F[_]] {
   def defaultContext: ExecutionContext
 
   def block[A](fa: F[A])(implicit F: Async[F]): F[A] =
-    for {
-      _ <- Async.shift(blockingContext)
-      aE <- fa.attempt
-      _ <- Async.shift(defaultContext)
-      a <- Applicative[F].pure(aE).rethrow
-    } yield a
+    F.bracket(Async.shift[F](blockingContext))(_ => fa)(_ => Async.shift[F](defaultContext))
 }
 
 object DualContext {

linebacker/src/main/scala/io/chrisdavenport/linebacker/Linebacker.scala

@@ -1,36 +1,44 @@
 package io.chrisdavenport.linebacker
 
-import cats._
-import cats.implicits._
-import cats.effect.Async
+import cats.effect._
 import java.util.concurrent.ExecutorService
 import scala.concurrent.ExecutionContext
 
 trait Linebacker[F[_]] {
 
-  def blockingPool: ExecutionContext
+  def blockingContext: ExecutionContext
+
+  @deprecated("0.2.0", "Use blockingContext instead.")
+  private[linebacker] def blockingPool: ExecutionContext = blockingContext
+
+  @deprecated("0.2.0", "Use blockShift instead.")
+  final private[linebacker] def block[A](
+      fa: F[A])(implicit F: Async[F], ec: ExecutionContext): F[A] =
+    blockShift(fa)(F, ec)
 
   /**
    * Attempts to Run the Given `F[A]` on the blocking pool.
    * Then shifts back to the given implicit execution context
    * after the Async `F[A]` is evaluated.
    */
-  final def block[A](fa: F[A])(implicit F: Async[F], ec: ExecutionContext): F[A] =
-    for {
-      _ <- Async.shift(blockingPool)
-      eA <- fa.attempt
-      _ <- Async.shift(ec)
-      a <- Applicative[F].pure(eA).rethrow
-    } yield a
+  final def blockShift[A](fa: F[A])(implicit F: Async[F], ec: ExecutionContext): F[A] =
+    F.bracket(Async.shift[F](blockingContext))(_ => fa)(_ => Async.shift[F](ec))
+
+  /**
+   * Attempts to Run the Given `F[A]` on the blocking pool.
+   * Then shifts back to the F for the timer.
+   */
+  final def blockTimer[A](fa: F[A])(implicit F: Async[F], timer: Timer[F]): F[A] =
+    F.bracket(Async.shift[F](blockingContext))(_ => fa)(_ => timer.shift)
 }
 
 object Linebacker {
   def apply[F[_]](implicit ev: Linebacker[F]): Linebacker[F] = ev
 
   def fromExecutorService[F[_]](es: ExecutorService): Linebacker[F] = new Linebacker[F] {
-    def blockingPool = ExecutionContext.fromExecutorService(es)
+    def blockingContext = ExecutionContext.fromExecutorService(es)
   }
   def fromExecutionContext[F[_]](ec: ExecutionContext): Linebacker[F] = new Linebacker[F] {
-    def blockingPool = ec
+    def blockingContext = ec
   }
 }

linebacker/src/main/scala/io/chrisdavenport/linebacker/Quarterback.scala

@@ -1,6 +1,5 @@
 package io.chrisdavenport.linebacker
 
-import cats._
 import cats.effect.Async
 import cats.implicits._
 import scala.concurrent.ExecutionContext
@@ -23,11 +22,8 @@ trait Quarterback[F[_], K] {
   def fleaFlicker[A](fa: F[A], initial: K, end: K)(implicit F: Async[F]): F[A] =
     for {
       iEC <- select(initial)
-      _ <- Async.shift(iEC)
-      aE <- fa.attempt
-      eEC <- select(end)
-      _ <- Async.shift(eEC)
-      a <- Applicative[F].pure(aE).rethrow
+      endEC <- select(end)
+      a <- Async[F].bracket(Async.shift[F](iEC))(_ => fa)(_ => Async.shift(endEC))
     } yield a
 }
 object Quarterback {

linebacker/src/main/scala/io/chrisdavenport/linebacker/contexts/Executors.scala

@@ -1,8 +1,7 @@
 package io.chrisdavenport.linebacker.contexts
 
-import cats.effect.Sync
+import cats.effect.{Resource, Sync}
 import cats.implicits._
-import fs2.Stream
 import java.util.concurrent.atomic.AtomicLong
 import java.util.concurrent.{ExecutorService, Executors => E, ForkJoinPool, ThreadFactory}
 
@@ -18,16 +17,16 @@ object Executors {
    * resource with the same number of threads.
    */
   def fixedPool[F[_]: Sync](n: Int) =
-    streamExecutorService(fixedPoolExecutorUnsafe(n))
+    executorServiceResource(fixedPoolExecutorUnsafe(n))
 
   /**
    * Constructs an unbound thread pool that will create
    * a new thread for each submitted job. This is useful
    * if you have a construct that is blocking but
    * self-manages the number of threads you can consume.
    */
-  def unbound[F[_]: Sync]: Stream[F, ExecutorService] =
-    streamExecutorService(unboundExecutorUnsafe)
+  def unbound[F[_]: Sync]: Resource[F, ExecutorService] =
+    executorServiceResource(unboundExecutorUnsafe)
 
   /**
    * A work stealing pool is often a useful blocking
@@ -39,20 +38,18 @@ object Executors {
    * requests or other work.
    */
   def workStealingPool[F[_]: Sync](n: Int) =
-    streamExecutorService(workStealingPoolUnsafe(n))
+    executorServiceResource(workStealingPoolUnsafe(n))
 
   /**
    * Default Pool For Scala, optimized for forked work and then returning to a
    * main pool, generally ideal for your main event loop.
    */
   def forkJoinPool[F[_]: Sync](n: Int) =
-    streamExecutorService(forkJoinPoolUnsafe(n))
+    executorServiceResource(forkJoinPoolUnsafe(n))
 
-  private def streamExecutorService[F[_]: Sync](f: F[ExecutorService]): Stream[F, ExecutorService] =
-    Stream.bracket(f)(
-      _.pure[Stream[F, ?]],
-      es => Sync[F].delay(es.shutdownNow).void
-    )
+  private def executorServiceResource[F[_]: Sync](
+      f: F[ExecutorService]): Resource[F, ExecutorService] =
+    Resource.make[F, ExecutorService](f)(es => Sync[F].delay(es.shutdownNow).void)
 
   object unsafe {
     def unboundExecutorUnsafe[F[_]: Sync]: F[ExecutorService] = Sync[F].delay {

linebacker/src/test/scala/io/chrisdavenport/linebacker/LinebackerSpec.scala

@@ -1,9 +1,8 @@
 package io.chrisdavenport.linebacker
 
 import org.specs2._
-import cats.effect.IO
+import cats.effect._
 import cats.implicits._
-import fs2.Stream
 import java.lang.Thread
 import scala.concurrent.ExecutionContext
 import java.util.concurrent.atomic.AtomicLong
@@ -17,17 +16,15 @@ class LinebackerSpec extends Spec {
   """
 
   def runsOnLinebacker = {
-    E.unbound[IO]
+    val testRun = E
+      .unbound[IO]
       .map(Linebacker.fromExecutorService[IO])
-      .flatMap { implicit linebacker =>
+      .use { implicit linebacker =>
         import scala.concurrent.ExecutionContext.Implicits.global
-        Stream.eval(
-          Linebacker[IO].block(IO(Thread.currentThread().getName))
-        )
+        Linebacker[IO].blockShift(IO(Thread.currentThread().getName))
       }
-      .compile
-      .last
-      .unsafeRunSync must_== Some("linebacker-thread-0")
+
+    testRun.unsafeRunSync must_=== "linebacker-thread-0"
   }
 
   def runsOffLinebackerAfterwards = {
@@ -46,14 +43,10 @@ class LinebackerSpec extends Spec {
 
     implicit val linebacker = Linebacker.fromExecutionContext[IO](ec)
 
-    Stream
-      .eval(
-        Linebacker[IO].block(IO.unit) *>
-          IO(Thread.currentThread().getName)
-          <* IO(executor.shutdownNow)
-      )
-      .compile
-      .last
-      .unsafeRunSync must_== Some("test-ec-1")
+    val testRun = Linebacker[IO].blockShift(IO.unit) *>
+      IO(Thread.currentThread().getName) <*
+      IO(executor.shutdownNow)
+
+    testRun.unsafeRunSync must_=== "test-ec-1"
   }
 }

project/build.properties

@@ -1 +1 @@
-sbt.version=1.1.5
+sbt.version=1.1.6

version.sbt

@@ -1 +1 @@
-version in ThisBuild := "0.1.1-SNAPSHOT"
+version in ThisBuild := "0.2.0-SNAPSHOT"