Follow 'em!

build.sbt

@@ -30,17 +30,20 @@ val providedDependencies = Seq(
 ).map(_ % "provided")
 
 val testDependencies = Seq(
-  "org.scalatest" %% "scalatest" % "3.1.0"
+  "org.scalamock" %% "scalamock" % "4.4.0",
+  "org.scalatest" %% "scalatest" % "3.1.0",
 ).map(_ % "test")
 
 val dependenciesToEmbed = Seq(
   "org.flywaydb" % "flyway-core" % "5.2.4",
-  "org.typelevel" %% "cats-core" % "2.0.0",
-  "org.typelevel" %% "cats-effect" % "2.0.0",
+  "org.typelevel" %% "cats-core" % "2.1.0",
+  "org.typelevel" %% "cats-effect" % "2.1.0",
   "eu.timepit" %% "refined" % "0.9.10",
   "com.beachape" %% "enumeratum" % "1.5.13"
 )
 
+addCompilerPlugin("org.typelevel" %% "kind-projector" % "0.11.0" cross CrossVersion.full)
+
 // localDependenciesはprovidedとして扱い、jarに埋め込まない
 assemblyExcludedJars in assembly := {
   (fullClasspath in assembly).value
@@ -87,6 +90,7 @@ lazy val root = (project in file("."))
     scalacOptions ++= Seq(
       "-encoding", "utf8",
       "-unchecked",
+      "-language:higherKinds",
       "-deprecation",
       "-Ypatmat-exhaust-depth", "320",
     ),

src/main/scala/com/github/unchama/buildassist/PlayerInventoryListener.scala

@@ -17,7 +17,7 @@ import scala.collection.mutable
 class PlayerInventoryListener extends Listener {
   val playerMap: mutable.HashMap[UUID, PlayerData] = BuildAssist.playermap
 
-  import com.github.unchama.seichiassist.concurrent.PluginExecutionContexts.sync
+  import com.github.unchama.seichiassist.concurrent.PluginExecutionContexts.syncShift
   import com.github.unchama.targetedeffect._
   import com.github.unchama.util.syntax.Nullability.NullabilityExtensionReceiver
 
@@ -64,7 +64,7 @@ class PlayerInventoryListener extends Listener {
 			 */
       if (itemstackcurrent.getType == Material.SKULL_ITEM) {
         //ホームメニューへ帰還
-        import com.github.unchama.seichiassist.concurrent.PluginExecutionContexts.{layoutPreparationContext, sync}
+        import com.github.unchama.seichiassist.concurrent.PluginExecutionContexts.{layoutPreparationContext, syncShift}
 
         seichiassist.unsafe.runAsyncTargetedEffect(player)(
           sequentialEffect(

src/main/scala/com/github/unchama/buildassist/listener/PlayerLeftClickListener.scala

@@ -29,7 +29,7 @@ object PlayerLeftClickListener extends Listener {
       if (!hasStickOnMainHand || !actionWasOnMainHand) return
     }
 
-    import com.github.unchama.seichiassist.concurrent.PluginExecutionContexts.{layoutPreparationContext, sync}
+    import com.github.unchama.seichiassist.concurrent.PluginExecutionContexts.{layoutPreparationContext, syncShift}
 
     seichiassist.unsafe.runAsyncTargetedEffect(player)(
       sequentialEffect(

src/main/scala/com/github/unchama/buildassist/menu/BlockPlacementSkillMenu.scala

@@ -15,7 +15,7 @@ import org.bukkit.{Material, Sound}
 
 object BlockPlacementSkillMenu extends Menu {
 
-  import com.github.unchama.seichiassist.concurrent.PluginExecutionContexts.{layoutPreparationContext, sync}
+  import com.github.unchama.seichiassist.concurrent.PluginExecutionContexts.{layoutPreparationContext, syncShift}
   import com.github.unchama.targetedeffect._
   import com.github.unchama.targetedeffect.syntax._
   import menuinventory.syntax._

src/main/scala/com/github/unchama/buildassist/menu/BuildMainMenu.scala

@@ -17,7 +17,7 @@ import org.bukkit.{Material, Sound}
 object BuildMainMenu extends Menu {
 
   import com.github.unchama.menuinventory.slot.button.RecomputedButton
-  import com.github.unchama.seichiassist.concurrent.PluginExecutionContexts.{layoutPreparationContext, sync}
+  import com.github.unchama.seichiassist.concurrent.PluginExecutionContexts.{layoutPreparationContext, syncShift}
   import com.github.unchama.targetedeffect._
   import com.github.unchama.targetedeffect.player.PlayerEffects._
   import com.github.unchama.targetedeffect.syntax._

src/main/scala/com/github/unchama/concurrent/BukkitSyncIOShift.scala

@@ -0,0 +1,50 @@
+package com.github.unchama.concurrent
+
+import cats.effect.{ContextShift, IO}
+import org.bukkit.Bukkit
+import org.bukkit.plugin.java.JavaPlugin
+
+import scala.concurrent.ExecutionContext
+
+/**
+ * Bukkitのメインスレッドへ実行をシフトするような `ContextShift[IO]` の実装。
+ *
+ * Bukkitのスケジューラにタスクを投げるような `ExecutionContext` から
+ * `ContextShift[IO]` を作ることはできるが、この実装だと一つ問題がある。
+ * 具体的には、`IO` のキャンセルがタスクの実行がキューされるまでしか可能でなく、
+ * 一度キューされたら実行が始まるのを待ち続けないといけない点にある。
+ * cats-effectの言葉で言えば、キャンセルが可能な非同期境界がキュー以前までしか無い。
+ *
+ * この問題点はBukkitと組み合わせると次の不都合がある。
+ * 例えば、Bukkitサーバーが停まるまでに必ず終了（正常終了かキャンセル）
+ * されなければならないプログラムがあったとする。
+ * また、このプログラムは途中で同期スレッドに実行をシフトするロジックを含むとする。
+ *
+ * サーバーの停止時、 `onDisable` にてこのプログラムの `Fiber` をcancelする必要がある。
+ * しかし、 `onDisable` が呼ばれた後はサーバーティックはもう来ない。
+ * よって、もし `onDisable` が同期スレッドへのシフトがキューされた後、
+ * かつ同期スレッドでのシフト後の処理が走り出す前に呼び出されてしまった場合、
+ * プログラムをキャンセルする手立てがない。デッドロックのような状況が発生し、
+ * リソースの安全な解放ができない可能性がある。
+ *
+ * `BukkitSyncIOShift` はこの問題を解決するため、
+ * shiftの実行時点でキャンセルが入ったときはBukkitのスケジューラ上で
+ * キューしたタスクもろともキャンセルするような実装になっている。
+ */
+sealed trait BukkitSyncIOShift extends ContextShift[IO]
+
+object BukkitSyncIOShift {
+  def apply()(implicit hostPlugin: JavaPlugin): BukkitSyncIOShift =
+    new BukkitSyncIOShift {
+      override def shift: IO[Unit] =
+        IO.cancelable { cb =>
+          val run = new Runnable { def run(): Unit = cb(Right(())) }
+          val scheduledTask = Bukkit.getScheduler.runTask(hostPlugin, run)
+
+          IO(scheduledTask.cancel())
+        }
+
+      override def evalOn[A](ec: ExecutionContext)(fa: IO[A]): IO[A] =
+        IO.shift(ec).bracket(_ => fa)(_ => shift)
+    }
+}

src/main/scala/com/github/unchama/concurrent/Execution.scala

@@ -7,9 +7,9 @@ object Execution {
    * 与えられた`IO`をサーバーメインスレッドで実行するように予約する。
    * @return
    */
-  def onServerMainThread(program: IO[Any])(implicit context: BukkitSyncExecutionContext): IO[Unit] = {
+  def onServerMainThread(program: IO[Any])(implicit sync: BukkitSyncIOShift): IO[Unit] = {
     val asyncProgram = for {
-      _ <- IO.shift(context)
+      _ <- sync.shift
       _ <- program
     } yield ()
 

src/main/scala/com/github/unchama/generic/OptionTExtra.scala

@@ -0,0 +1,12 @@
+package com.github.unchama.generic
+
+import cats.data.OptionT
+import cats.{Applicative, ~>}
+
+object OptionTExtra {
+  /**
+   * `failCondition` が true のとき失敗するような計算を返す。
+   */
+  def failIf[F[_]: Applicative](failCondition: Boolean): OptionT[F, Unit] =
+    OptionT.fromOption[F](Option.unless(failCondition)(()))
+}

src/main/scala/com/github/unchama/generic/effect/ConcurrentExtra.scala

@@ -0,0 +1,37 @@
+package com.github.unchama.generic.effect
+
+import cats.effect.concurrent.Deferred
+import cats.effect.{CancelToken, Concurrent}
+
+object ConcurrentExtra {
+  import cats.implicits._
+
+  /**
+   * `f` を並行的に開始し、 `f` の計算をキャンセルする計算を`f`に渡してから結果をawaitする計算を返す。
+   *
+   * `f` 内で渡された `CancelToken[F]` を実行した際の動作は未定義となる。
+   * 実際、`f` 内のキャンセルはそれ自身の終了をブロックしながらawaitするため、
+   * ハングすることが予想される。
+   */
+  def withSelfCancellation[F[_]: Concurrent, A](f: CancelToken[F] => F[A]): F[A] =
+    for {
+      //  [start `awaitToken >>= f`]--
+      //             |               |
+      //      [fill promise]         |
+      //             |--->[await promise completion]
+      //             |               |
+      //             |        [`f` yielding a]
+      //             |               |
+      //     [await completion]<------
+      //             |
+      // yield a <----
+
+      tokenPromise <- Deferred[F, CancelToken[F]]
+
+      fiber <- Concurrent[F].start(tokenPromise.get.flatMap(f))
+
+      _ <- tokenPromise.complete(fiber.cancel)
+
+      a <- fiber.join
+    } yield a
+}