Implement cancellation

3 files changed, 48 insertions, 18 deletions

diff --git a/build b/build
index d080658..d4f92c1 100755
--- a/build
+++ b/build
@@ -1,4 +1,17 @@
 #!/bin/sh
 
-curl -X POST -H "Content-Type: application/json" http://localhost:9000/api/submit \
-     -d "{\"url\": \"$1\"}"
+case "$1" in
+    start)
+	shift
+	curl -i -X POST -H "Content-Type: application/json" "http://localhost:9000/api/submit" \
+	     -d "{\"url\": \"$1\"}"
+	;;
+    cancel)
+	shift
+	curl -i -X POST "http://localhost:9000/api/$1/cancel"
+	;;
+    *)
+	echo "invalid command $1" >&2
+	exit 2
+	;;
+esac
diff --git a/crashbox-server/src/main/scala/io/crashbox/ci/HttpApi.scala b/crashbox-server/src/main/scala/io/crashbox/ci/HttpApi.scala
index 7cb76b0..71a4f5b 100644
--- a/crashbox-server/src/main/scala/io/crashbox/ci/HttpApi.scala
+++ b/crashbox-server/src/main/scala/io/crashbox/ci/HttpApi.scala
@@ -32,7 +32,7 @@ trait HttpApi { self: Core with Schedulers with StreamStore =>
   }
   import Protocol._
 
-  implicit val toResponseMarshaller: ToResponseMarshaller[Src[BuildState, Any]] =
+  implicit val toResponseMarshaller: ToResponseMarshaller[Src[String, Any]] =
     Marshaller.opaque { items =>
       val data = items.map(item => ChunkStreamPart(item.toString + "\n"))
       HttpResponse(
@@ -41,22 +41,29 @@ trait HttpApi { self: Core with Schedulers with StreamStore =>
 
   def httpApi: Route = pathPrefix(endpoint) {
     path("submit") {
-      entity(as[Request]) { req =>
-        val source = Src
-          .queue[BuildState](100, OverflowStrategy.fail)
-          .mapMaterializedValue { q =>
-            start(
-              req.buildId,
-              new URL(req.url),
-              () => saveStream(req.buildId),
-              state => q.offer(state)
-            )
-          }
-
-        complete(source)
+      post {
+        entity(as[Request]) { req =>
+          val source = Src
+            .queue[String](100, OverflowStrategy.fail)
+            .mapMaterializedValue { q =>
+              q.offer(s"Build ID: ${req.buildId}")
+              start(
+                req.buildId,
+                new URL(req.url),
+                () => saveStream(req.buildId),
+                state => q.offer(state.toString)
+              )
+            }
+          complete(source)
+        }
+      }
+    } ~
+      path(Segment / "cancel") { buildId =>
+        post {
+          cancel(buildId)
+          complete(204 -> s"Cancelled $buildId")
+        }
       }
-
-    }
   }
 
 }
diff --git a/crashbox-server/src/main/scala/io/crashbox/ci/Schedulers.scala b/crashbox-server/src/main/scala/io/crashbox/ci/Schedulers.scala
index 7632d95..3600fc1 100644
--- a/crashbox-server/src/main/scala/io/crashbox/ci/Schedulers.scala
+++ b/crashbox-server/src/main/scala/io/crashbox/ci/Schedulers.scala
@@ -119,6 +119,7 @@ trait Schedulers extends { self: Core with Source with Builders with Parsers =>
       out: () => OutputStream,
       update: BuildState => Unit
   ) extends SchedulerCommand
+  private case class CancelBuild(buildId: String) extends SchedulerCommand
 
   class Scheduler extends Actor {
 
@@ -137,6 +138,11 @@ trait Schedulers extends { self: Core with Source with Builders with Parsers =>
             runningBuilds += sb.buildId -> buildManager
         }
 
+      case CancelBuild(id) =>
+        runningBuilds.get(id).foreach { builder =>
+          context.stop(builder)
+        }
+
       case Terminated(buildManager) =>
         //TODO use a more efficient data structure
         runningBuilds.find(_._2 == buildManager).foreach {
@@ -157,4 +163,8 @@ trait Schedulers extends { self: Core with Source with Builders with Parsers =>
     scheduler ! ScheduleBuild(buildId, url, out, update)
   }
 
+  def cancel(buildId: String): Unit = {
+    scheduler ! CancelBuild(buildId)
+  }
+
 }