shutdown dispatcher

blaze-client/src/test/scala/org/http4s/client/blaze/ClientTimeoutSuite.scala

@@ -19,20 +19,21 @@ package client
 package blaze
 
 import cats.effect._
-import cats.effect.std.{Dispatcher, Queue}
-import cats.syntax.all._
+import cats.effect.std.Queue
 import fs2.Stream
+
 import java.io.IOException
 import java.nio.ByteBuffer
 import java.nio.charset.StandardCharsets
 import org.http4s.blaze.pipeline.HeadStage
 import org.http4s.blaze.util.TickWheelExecutor
 import org.http4s.blazecore.{QueueTestHead, SeqTestHead, SlowTestHead}
+import org.http4s.testing.DispatcherIOFixture
+
 import scala.concurrent.TimeoutException
 import scala.concurrent.duration._
 
-class ClientTimeoutSuite extends Http4sSuite {
-  val dispatcher = Dispatcher[IO].allocated.map(_._1).unsafeRunSync()
+class ClientTimeoutSuite extends Http4sSuite with DispatcherIOFixture {
 
   def fixture = ResourceFixture(
     Resource.make(IO(new TickWheelExecutor(tick = 50.millis)))(tickWheel =>
@@ -53,7 +54,7 @@ class ClientTimeoutSuite extends Http4sSuite {
       chunkBufferMaxSize = 1024 * 1024,
       parserMode = ParserMode.Strict,
       userAgent = None,
-      dispatcher = dispatcher
+      dispatcher = dispatcher()
     )
 
   private def mkBuffer(s: String): ByteBuffer =

blaze-client/src/test/scala/org/http4s/client/blaze/Http1ClientStageSuite.scala

@@ -19,20 +19,21 @@ package client
 package blaze
 
 import cats.effect._
-import cats.effect.std.{Dispatcher, Queue}
-import cats.syntax.all._
+import cats.effect.std.Queue
 import fs2.Stream
+
 import java.nio.ByteBuffer
 import java.nio.charset.StandardCharsets
 import org.http4s.blaze.pipeline.LeafBuilder
 import org.http4s.blazecore.{QueueTestHead, SeqTestHead}
 import org.http4s.client.blaze.bits.DefaultUserAgent
 import org.http4s.headers.`User-Agent`
+import org.http4s.testing.DispatcherIOFixture
 import org.typelevel.ci.CIString
+
 import scala.concurrent.duration._
 
-class Http1ClientStageSuite extends Http4sSuite {
-  val dispatcher = Dispatcher[IO].allocated.map(_._1).unsafeRunSync()
+class Http1ClientStageSuite extends Http4sSuite with DispatcherIOFixture {
 
   val trampoline = org.http4s.blaze.util.Execution.trampoline
 
@@ -62,7 +63,7 @@ class Http1ClientStageSuite extends Http4sSuite {
       chunkBufferMaxSize = 1024,
       parserMode = ParserMode.Strict,
       userAgent = userAgent,
-      dispatcher = dispatcher
+      dispatcher = dispatcher()
     )
 
   private def mkBuffer(s: String): ByteBuffer =

blaze-server/src/test/scala/org/http4s/server/blaze/Http1ServerStageSpec.scala

@@ -47,9 +47,13 @@ class Http1ServerStageSpec extends Http4sSpec with AfterAll {
 
   val tickWheel = new TickWheelExecutor()
 
-  val dispatcher = Dispatcher[IO].allocated.map(_._1).unsafeRunSync()
+  val dispatcherAndShutdown = Dispatcher[IO].allocated.unsafeRunSync()
+  val dispatcher = dispatcherAndShutdown._1
 
-  def afterAll() = tickWheel.shutdown()
+  def afterAll() = {
+    tickWheel.shutdown()
+    dispatcherAndShutdown._2.unsafeRunSync()
+  }
 
   def makeString(b: ByteBuffer): String = {
     val p = b.position()

testing/src/test/scala/org/http4s/testing/DispatcherIOFixture.scala

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2016 http4s.org
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.http4s.testing
+
+import cats.effect.IO
+import cats.effect.std.Dispatcher
+import org.http4s.Http4sSuite
+
+trait DispatcherIOFixture { this: Http4sSuite =>
+
+  val dispatcher: Fixture[Dispatcher[IO]] = new Fixture[Dispatcher[IO]]("dispatcher") {
+    var dispatcher: Dispatcher[IO] = _
+    var shutdown: IO[Unit] = _
+    override def apply(): Dispatcher[IO] = dispatcher
+
+    override def beforeAll(): Unit = {
+      val dispatcherAndShutdown = Dispatcher[IO].allocated.unsafeRunSync()
+      dispatcher = dispatcherAndShutdown._1
+      shutdown = dispatcherAndShutdown._2
+    }
+
+    override def afterAll(): Unit =
+      shutdown.unsafeRunSync()
+  }
+
+}