fix: Added OperationQueue to TaskGraph.

src/commands/auto-reload.ts

@@ -1,15 +1,23 @@
+/*
+ * Copyright (C) 2018 Garden Technologies, Inc. <[email protected]>
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/.
+ */
+
 import { keys, values } from "lodash"
 import { Command } from "./base"
 import { Module } from "../types/module"
-import { GardenContext } from "../context"
 import { FSWatcher } from "../fs-watcher"
+import { PluginContext } from "../plugin-context"
 import { BuildTask } from "../tasks/build"
 import { DeployTask } from "../tasks/deploy"
 import { registerCleanupFunction, sleep } from "../util"
 
 export type AutoReloadDependants = { [key: string]: Set<Module> }
 
-async function registerAutoReloadWatches(ctx: GardenContext): Promise<FSWatcher | null> {
+async function registerAutoReloadWatches(ctx: PluginContext): Promise<FSWatcher | null> {
   const allModules = values(await ctx.getModules())
   const modules = allModules.filter((m) => !m.skipAutoReload)
 
@@ -25,7 +33,7 @@ async function registerAutoReloadWatches(ctx: GardenContext): Promise<FSWatcher
 
   const autoReloadDependants = await computeAutoReloadDependants(modules)
 
-  const watcher = new FSWatcher(ctx)
+  const watcher = new FSWatcher(ctx.projectRoot)
   watcher.watchModules(modules, "addTasksForAutoReload/",
     async (changedModule, _) => {
       ctx.log.info({ msg: `files changed for module ${changedModule.name}` })
@@ -50,7 +58,7 @@ export async function computeAutoReloadDependants(modules: Module[]):
   return dependants
 }
 
-export async function addTasksForAutoReload(ctx: GardenContext, module: Module, dependants: AutoReloadDependants) {
+export async function addTasksForAutoReload(ctx: PluginContext, module: Module, dependants: AutoReloadDependants) {
   const serviceNames = keys(module.services || {})
 
   if (serviceNames.length === 0) {
@@ -74,7 +82,7 @@ export class AutoReloadCommand extends Command {
   name = "autoreload"
   help = "Auto-reload modules when sources change"
 
-  async action(ctx: GardenContext): Promise<void> {
+  async action(ctx: PluginContext): Promise<void> {
     const watcher = await registerAutoReloadWatches(ctx)
 
     if (!watcher) {

src/fs-watcher.ts

@@ -1,9 +1,16 @@
+/*
+ * Copyright (C) 2018 Garden Technologies, Inc. <[email protected]>
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/.
+ */
+
 import { map as bluebirdMap } from "bluebird"
 import { Client } from "fb-watchman"
 import { keyBy } from "lodash"
 import { resolve } from "path"
 import { Module } from "./types/module"
-import { GardenContext } from "./context"
 
 export type CapabilityOptions = { required?: string[], optional?: string[] }
 export type CapabilityResponse = { error: Error, response: { capabilities: { string: boolean } } }
@@ -22,12 +29,10 @@ export type SubscriptionResponse = {
 }
 
 export class FSWatcher {
-  readonly ctx: GardenContext
   private readonly client
   private capabilityCheckComplete: boolean
 
-  constructor(ctx: GardenContext) {
-    this.ctx = ctx
+  constructor(private projectRoot: string) {
     this.client = new Client()
     this.capabilityCheckComplete = false
   }
@@ -53,6 +58,7 @@ export class FSWatcher {
     })
   }
 
+  // WIP
   async watchModules(modules: Module[], subscriptionPrefix: string,
     changeHandler: (Module, SubscriptionResponse) => Promise<void>) {
     if (!this.capabilityCheckComplete) {
@@ -63,10 +69,17 @@ export class FSWatcher {
 
     await bluebirdMap(modules || [], async (module) => {
       const subscriptionKey = FSWatcher.subscriptionKey(subscriptionPrefix, module)
-      const modulePath = resolve(this.ctx.projectRoot, module.path)
+      const modulePath = resolve(this.projectRoot, module.path)
+
       const result = await this.command(["watch-project", modulePath])
 
-      const subscriptionRequest = {}
+      // console.log("watching", modulePath)
+
+      const subscriptionRequest = {
+        // expression: ["anyof",
+        //   ["dirname", modulePath, ["depth", "ge", 0]]
+        // ]
+      }
 
       await this.command([
         "subscribe",
@@ -76,7 +89,7 @@ export class FSWatcher {
     })
 
     this.on("subscription", async (response) => {
-      console.log("file changed:", response)
+      // console.log("file changed:", response)
       const changedModule = modulesBySubscriptionKey[response.subscription]
       if (!changedModule) {
         console.log("no module found for changed file, skipping auto-rebuild")

src/task-graph.ts

@@ -9,7 +9,6 @@
 import * as Bluebird from "bluebird"
 import chalk from "chalk"
 import { pick } from "lodash"
-import { GardenContext } from "./context"
 import { Task, TaskDefinitionError } from "./types/task"
 
 import { EntryStyle, LogSymbolType } from "./logger/types"
@@ -48,16 +47,22 @@ export class TaskGraph {
   private inProgress: TaskNodeMap
   private logEntryMap: LogEntryMap
 
+  private opQueue: OperationQueue
+
   constructor(private ctx: PluginContext, private concurrency: number = DEFAULT_CONCURRENCY) {
     this.roots = new TaskNodeMap()
     this.index = new TaskNodeMap()
     this.inProgress = new TaskNodeMap()
+    this.opQueue = new OperationQueue(this)
     this.logEntryMap = {}
   }
 
-  async addTask(task: Task) {
-    // TODO: Detect circular dependencies.
+  addTask(task: Task): Promise<any> {
+    return this.opQueue.request({ type: "addTask", task })
+  }
 
+  async addTaskInternal(task: Task) {
+    // TODO: Detect circular dependencies.
     const predecessor = this.getPredecessor(task)
     let node = this.getNode(task)
 
@@ -96,10 +101,15 @@ export class TaskGraph {
     return existing || new TaskNode(task)
   }
 
+  processTasks(): Promise<TaskResults> {
+    return this.opQueue.request({ type: "processTasks" })
+  }
+
   /*
     Process the graph until it's complete
    */
-  async processTasks(): Promise<TaskResults> {
+  async processTasksInternal(): Promise<TaskResults> {
+
     const _this = this
     let results: TaskResults = {}
 
@@ -182,7 +192,7 @@ export class TaskGraph {
     const nodeDependencies = node.getDependencies()
     for (const d of nodeDependencies) {
       const dependant = this.getPredecessor(d.task) || d
-      await this.addTask(dependant.task)
+      await this.addTaskInternal(dependant.task)
       dependant.addDependant(node)
     }
   }
@@ -349,3 +359,81 @@ class TaskNode {
     return await this.task.process(dependencyResults)
   }
 }
+
+// TODO: Add more typing to this class.
+
+/*
+  Used by TaskGraph to prevent race conditions e.g. when calling addTask or
+  processTasks.
+*/
+class OperationQueue {
+  queue: object[]
+  draining: boolean
+
+  constructor(private taskGraph: TaskGraph) {
+    this.queue = []
+    this.draining = false
+  }
+
+  request(opRequest): Promise<any> {
+    let findFn
+
+    switch (opRequest.type) {
+
+      case "addTask":
+        findFn = (o) => o.type === "addTask" && o.task.getBaseKey() === opRequest.task.getBaseKey()
+        break
+
+      case "processTasks":
+        findFn = (o) => o.type === "processTasks"
+        break
+    }
+
+    const existingOp = this.queue.find(findFn)
+
+    const prom = new Promise((resolver) => {
+      if (existingOp) {
+        existingOp["resolvers"].push(resolver)
+      } else {
+        this.queue.push({ ...opRequest, resolvers: [resolver] })
+      }
+    })
+
+    if (!this.draining) {
+      this.process()
+    }
+
+    return prom
+  }
+
+  async process() {
+    this.draining = true
+    const op = this.queue.shift()
+
+    if (!op) {
+      this.draining = false
+      return
+    }
+
+    switch (op["type"]) {
+
+      case "addTask":
+        const task = op["task"]
+        await this.taskGraph.addTaskInternal(task)
+        for (const resolver of op["resolvers"]) {
+          resolver()
+        }
+        break
+
+      case "processTasks":
+        const results = await this.taskGraph.processTasksInternal()
+        for (const resolver of op["resolvers"]) {
+          resolver(results)
+        }
+        break
+    }
+
+    this.process()
+  }
+
+}

src/tasks/deploy.ts

@@ -10,7 +10,6 @@ import { LogEntry } from "../logger"
 import { PluginContext } from "../plugin-context"
 import { BuildTask } from "./build"
 import { values } from "lodash"
-import { Service } from "../types/service"
 import { Task } from "../types/task"
 import {
   Service,

src/tasks/push.ts

@@ -7,12 +7,12 @@
  */
 
 import chalk from "chalk"
-import { Task } from "../task-graph"
 import { PluginContext } from "../plugin-context"
 import { BuildTask } from "./build"
 import { Module } from "../types/module"
 import { EntryStyle } from "../logger/types"
 import { PushResult } from "../types/plugin"
+import { Task } from "../types/task"
 
 export class PushTask<T extends Module<any>> extends Task {
   type = "push"
@@ -31,7 +31,7 @@ export class PushTask<T extends Module<any>> extends Task {
     return [new BuildTask(this.ctx, this.module, this.forceBuild)]
   }
 
-  getKey() {
+  getName() {
     // TODO: Include version in the task key (may need to make this method async).
     return this.module.name
   }

src/tasks/test.ts

@@ -7,10 +7,8 @@
  */
 
 import { PluginContext } from "../plugin-context"
-import { Task } from "../task-graph"
 import { Module, TestSpec } from "../types/module"
 import { BuildTask } from "./build"
-import { TestResult } from "../types/plugin"
 import { DeployTask } from "./deploy"
 import { TestResult } from "../types/plugin"
 import { Task } from "../types/task"

src/types/task.ts

@@ -1,3 +1,11 @@
+/*
+ * Copyright (C) 2018 Garden Technologies, Inc. <[email protected]>
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/.
+ */
+
 import { TaskResults } from "../task-graph"
 import { v1 as uuidv1 } from "uuid"
 

test/src/build-dir.ts

@@ -4,9 +4,7 @@ import { pathExists, readdir } from "fs-extra"
 import { expect } from "chai"
 import { values } from "lodash"
 import { BuildTask } from "../../src/tasks/build"
-import {
-  makeTestGarden,
-} from "../helpers"
+import { makeTestGarden, } from "../helpers"
 
 /*
   Module dependency diagram for test-project-build-products