Updates to fix coercion errors. DSDEEPB-727

src/main/resources/reference.conf

@@ -30,7 +30,7 @@ database {
 
   main {
     hsqldb {
-      url = "jdbc:hsqldb:mem:tempdb;shutdown=false"
+      url = "jdbc:hsqldb:mem:${slick.uniqueSchema};shutdown=false"
       driver = "org.hsqldb.jdbcDriver"
       slick.driver = "slick.driver.HsqldbDriver"
       slick.createSchema = true

src/main/scala/cromwell/Main.scala

@@ -93,6 +93,8 @@ object Main extends App {
       case e: Exception =>
         println("Unable to process inputs")
         e.printStackTrace()
+    } finally {
+      workflowManagerSystem.shutdown() // Future[Unit], but give it a shot
     }
   }
 

src/main/scala/cromwell/binding/WdlNamespace.scala

@@ -421,5 +421,7 @@ case class WdlNamespace(ast: Ast, source: WdlSource, importResolver: ImportResol
     }
   }
 
+  override def toRawString = ???
+
   validate()
 }

src/main/scala/cromwell/binding/types/WdlBooleanType.scala

@@ -14,5 +14,6 @@ case object WdlBooleanType extends WdlType {
     case s: JsString if s.value.equalsIgnoreCase("false") => WdlBoolean.False
     case s: JsBoolean => WdlBoolean(s.value)
   }
-}
 
+  override def fromRawString(rawString: String) = WdlBoolean(rawString.toBoolean)
+}

src/main/scala/cromwell/binding/types/WdlExpressionType.scala

@@ -8,5 +8,6 @@ case object WdlExpressionType extends WdlType {
   override protected def coercion = {
     case s: String if s.startsWith("%expr:") => WdlExpression.fromString(s.replace("%expr:", ""))
   }
-}
 
+  override def fromRawString(rawString: String) = WdlExpression.fromString(rawString)
+}

src/main/scala/cromwell/binding/types/WdlFileType.scala

@@ -10,4 +10,6 @@ case object WdlFileType extends WdlType {
     case s: String => WdlFile(s)
     case s: JsString => WdlFile(s.value)
   }
+
+  override def fromRawString(rawString: String) = WdlFile(rawString)
 }

src/main/scala/cromwell/binding/types/WdlFloatType.scala

@@ -10,5 +10,6 @@ case object WdlFloatType extends WdlType {
     case d: Double => WdlFloat(d)
     case n: JsNumber => WdlFloat(n.value.doubleValue())
   }
-}
 
+  override def fromRawString(rawString: String) = WdlFloat(rawString.toFloat)
+}

src/main/scala/cromwell/binding/types/WdlIntegerType.scala

@@ -8,6 +8,8 @@ case object WdlIntegerType extends WdlType {
 
   override protected def coercion = {
     case i: Integer => WdlInteger(i)
-    case n: JsNumber => WdlInteger(n.value.intValue)
+    case n: JsNumber => WdlInteger(n.value.intValue())
   }
-}
+
+  override def fromRawString(rawString: String) = WdlInteger(rawString.toInt)
+}

src/main/scala/cromwell/binding/types/WdlNamespaceType.scala

@@ -3,4 +3,5 @@ package cromwell.binding.types
 case object WdlNamespaceType extends WdlType {
   override def toWdlString: String = "Namespace"
   override protected def coercion = ???
+  override def fromRawString(rawString: String) = ???
 }

src/main/scala/cromwell/binding/types/WdlObjectType.scala

@@ -4,4 +4,6 @@ case object WdlObjectType extends WdlType {
   val toWdlString: String = "Object"
 
   override protected def coercion = ???
+
+  override def fromRawString(rawString: String) = ???
 }

src/main/scala/cromwell/binding/types/WdlStringType.scala

@@ -10,4 +10,6 @@ case object WdlStringType extends WdlType {
     case s: String => WdlString(s)
     case s: JsString => WdlString(s.value)
   }
+
+  override def fromRawString(rawString: String) = WdlString(rawString)
 }

src/main/scala/cromwell/binding/types/WdlType.scala

@@ -28,6 +28,13 @@ trait WdlType {
   }
 
   def toWdlString: String
+
+  /**
+   * Converts rawString generated from WdlValue.toRawString back to a WdlValue.
+   * @param rawString Raw string generated by WdlValue.toRawString
+   * @return The WdlValue
+   */
+  def fromRawString(rawString: String): WdlValue
 }
 
 object WdlType {

src/main/scala/cromwell/binding/values/WdlObject.scala

@@ -4,4 +4,6 @@ import cromwell.binding.types.WdlObjectType
 
 case class WdlObject(value: Map[String, WdlValue]) extends WdlValue {
   val wdlType = WdlObjectType
+
+  override def toRawString = ???
 }

src/main/scala/cromwell/binding/values/WdlPrimitive.scala

@@ -1,8 +1,6 @@
 package cromwell.binding.values
 
-import scala.util.Try
-
 trait WdlPrimitive extends WdlValue {
   def asString: String
-  override def toRawString = Try(asString)
+  override def toRawString = asString
 }

src/main/scala/cromwell/binding/values/WdlValue.scala

@@ -26,5 +26,6 @@ trait WdlValue {
   def not: Try[WdlValue] = invalid(s"!$this")
   def unaryPlus: Try[WdlValue] = invalid(s"+$this")
   def unaryMinus: Try[WdlValue] = invalid(s"-$this")
-  def toRawString: Try[String] = Try(toString)
+  def toRawString: String = toString
+  def typeName: String = wdlType.getClass.getSimpleName
 }

src/main/scala/cromwell/engine/db/DataAccess.scala

@@ -5,11 +5,30 @@ import cromwell.binding.{Call, FullyQualifiedName, WdlJson, WdlSource}
 import cromwell.engine.backend.Backend
 import cromwell.engine.{SymbolStoreEntry, WorkflowId, WorkflowState}
 
-import scala.concurrent.Future
+import scala.concurrent.duration.Duration
+import scala.concurrent.{Await, Future}
 
 object DataAccess {
   def apply(): DataAccess = new slick.SlickDataAccess()
 
+  /**
+   * Creates a DataAccess instance, loans it to the function,
+   * and then attempts to shut down the instance.
+   *
+   * @param f Function to run on the data access.
+   * @tparam T Return type of the function.
+   * @return Result of calling the function with the dataAccess instance.
+   */
+  def withDataAccess[T](f: DataAccess => T): T = {
+    val dataAccess = DataAccess()
+    try {
+      f(dataAccess)
+    } finally {
+      // NOTE: shutdown result thrown away
+      Await.ready(dataAccess.shutdown(), Duration.Inf)
+    }
+  }
+
   // TODO PLEASE RENAME ME
   case class WorkflowInfo(workflowId: WorkflowId, wdlSource: WdlSource, wdlJson: WdlJson)
 }
@@ -54,4 +73,6 @@ trait DataAccess {
 
   def getExecutionStatuses(workflowId: WorkflowId): Future[Map[FullyQualifiedName, CallStatus]]
 
+  /** Shutdown. NOTE: Should (internally or explicitly) use AsyncExecutor.shutdownExecutor. */
+  def shutdown(): Future[Unit]
 }

src/main/scala/cromwell/engine/db/slick/SlickDataAccess.scala

@@ -5,10 +5,9 @@ import java.util.{Date, UUID}
 import javax.sql.rowset.serial.SerialClob
 
 import _root_.slick.util.ConfigExtensionMethods._
-import com.typesafe.config.Config
+import com.typesafe.config.{Config, ConfigFactory, ConfigValueFactory}
 import cromwell.binding._
 import cromwell.binding.types.WdlType
-import cromwell.binding.values.WdlValue
 import cromwell.engine._
 import cromwell.engine.backend.Backend
 import cromwell.engine.backend.local.LocalBackend
@@ -36,6 +35,32 @@ object SlickDataAccess {
   implicit class StringToClob(val str: String) extends AnyVal {
     def toClob: Clob = new SerialClob(str.toCharArray)
   }
+
+  implicit class ConfigWithUniqueSchema(val config: Config) extends AnyVal {
+    /**
+     * Modifies config.getString("url") to return a unique schema, if the original url contains the text
+     * "${slick.uniqueSchema}".
+     *
+     * This allows each instance of a SlickDataAccess object to use a clean, and different, in memory database.
+     *
+     * @return Config with ${slick.uniqueSchema} in url replaced with a unique string.
+     */
+    def withUniqueSchema: Config = {
+      val url = config.getString("url")
+      if (url.contains("${slick.uniqueSchema}")) {
+        // Config wasn't updating with a simple withValue/withFallback.
+        // So instead, do a bit of extra work to insert the generated schema name in the url.
+        val schema = UUID.randomUUID().toString
+        val newUrl = url.replaceAll("""\$\{slick\.uniqueSchema\}""", schema)
+        val origin = "url with slick.uniqueSchema=" + schema
+        val urlConfigValue = ConfigValueFactory.fromAnyRef(newUrl, origin)
+        val urlConfig = ConfigFactory.empty(origin).withValue("url", urlConfigValue)
+        urlConfig.withFallback(config)
+      } else {
+        config
+      }
+    }
+  }
 }
 
 class SlickDataAccess(databaseConfig: Config, val dataAccess: DataAccessComponent) extends DataAccess {
@@ -56,7 +81,7 @@ class SlickDataAccess(databaseConfig: Config, val dataAccess: DataAccessComponen
   import dataAccess.driver.api._
 
   // NOTE: if you want to refactor database is inner-class type: this.dataAccess.driver.backend.DatabaseFactory
-  val database = Database.forConfig("", databaseConfig)
+  val database = Database.forConfig("", databaseConfig.withUniqueSchema)
 
   // Possibly create the database
   {
@@ -70,11 +95,18 @@ class SlickDataAccess(databaseConfig: Config, val dataAccess: DataAccessComponen
     //   used in key specification without a key length
     //
     // Perhaps we'll use a more optimized data type for UUID's bytes in the future, as a FK, instead auto-inc cols
+    //
+    // The value `${slick.uniqueSchema}` may be used in the url, in combination with `slick.createSchema = true`, to
+    // generate unique schema instances that don't conflict.
+    //
+    // Otherwise, create one DataAccess and hold on to the reference.
     if (databaseConfig.getBooleanOr("slick.createSchema")) {
       Await.result(database.run(dataAccess.schema.create), Duration.Inf)
     }
   }
 
+  override def shutdown() = database.shutdown
+
   // Run action with an outer transaction
   private def runTransaction[R](action: DBIOAction[R, _ <: NoStream, _ <: Effect]): Future[R] = {
     database.run(action.transactionally)
@@ -127,7 +159,7 @@ class SlickDataAccess(databaseConfig: Config, val dataAccess: DataAccessComponen
         symbol.key.iteration.getOrElse(IterationNone),
         if (symbol.isInput) IoInput else IoOutput,
         symbol.wdlType.toWdlString,
-        symbol.wdlValue.map(_.toRawString.get))
+        symbol.wdlValue.map(_.toRawString))
     }
   }
 
@@ -341,8 +373,7 @@ class SlickDataAccess(databaseConfig: Config, val dataAccess: DataAccessComponen
         new SymbolStoreEntry(
           symbolStoreKey,
           wdlType,
-          symbolResult.wdlValue.map(wdlType.coerceRawValue(_).get))
-
+          symbolResult.wdlValue.map(wdlType.fromRawString))
       }
 
     } yield symbolStoreEntries
@@ -351,7 +382,7 @@ class SlickDataAccess(databaseConfig: Config, val dataAccess: DataAccessComponen
   }
 
   /** Should fail if a value is already set.  The keys in the Map are locally qualified names. */
-  override def setOutputs(workflowId: WorkflowId, call: Call, callOutputs: Map[String, WdlValue]): Future[Unit] = {
+  override def setOutputs(workflowId: WorkflowId, call: Call, callOutputs: WorkflowOutputs): Future[Unit] = {
 
     val action = for {
       workflowExecution <- dataAccess.workflowExecutionsByWorkflowExecutionUuid(workflowId.toString).result.head
@@ -360,11 +391,11 @@ class SlickDataAccess(databaseConfig: Config, val dataAccess: DataAccessComponen
           new Symbol(
             workflowExecution.workflowExecutionId.get,
             call.fullyQualifiedName,
-            call.fullyQualifiedName + "." + symbolLocallyQualifiedName,
+            symbolLocallyQualifiedName,
             IterationNone,
             IoOutput,
             wdlValue.wdlType.toWdlString,
-            Option(wdlValue.toRawString.get))
+            Option(wdlValue.toRawString))
       }
     } yield ()
 

src/main/scala/cromwell/server/CromwellServer.scala

@@ -7,7 +7,6 @@ import akka.pattern.ask
 import akka.util.Timeout
 import com.typesafe.config.ConfigFactory
 import com.wordnik.swagger.model.ApiInfo
-import cromwell.engine.db.DataAccess
 import cromwell.webservice.{CromwellApiService, CromwellApiServiceActor, SwaggerService}
 import spray.can.Http
 
@@ -18,7 +17,12 @@ import scala.util.{Failure, Success}
 // Note that as per the language specification, this is instantiated lazily and only used when necessary (i.e. server mode)
 object CromwellServer extends DefaultWorkflowManagerSystem {
   val conf = ConfigFactory.parseFile(new File("/etc/cromwell.conf"))
-  private lazy val realDataAccess = DataAccess()
+
+  // NOTE: Currently the this.dataAccess is passed in to this.workflowManagerActor.
+  // The actor could technically restart with the same instance of the dataAccess,
+  // So, we're not shutting down dataAccess during this.workflowManagerActor.postStop() nor this.service.postStop().
+  // Not sure otherwise when this server is really shutting down, so this.dataAccess currently never explicitly closed.
+  // Shouldn't be an issue unless perhaps test code tries to launch multiple servers and leaves dangling connections.
 
   val swaggerConfig = conf.getConfig("swagger")
   val swaggerService = new SwaggerService(
@@ -51,7 +55,4 @@ object CromwellServer extends DefaultWorkflowManagerSystem {
     case _ =>
       actorSystem.log.info("Cromwell service started...")
   }
-
-  override def dataAccess: DataAccess = realDataAccess
 }
-

src/main/scala/cromwell/server/DefaultWorkflowManagerSystem.scala

@@ -1,7 +1,4 @@
 package cromwell.server
 
-import cromwell.engine.db.DataAccess
-
 case class DefaultWorkflowManagerSystem() extends WorkflowManagerSystem {
-  def dataAccess: DataAccess = DataAccess()
 }

src/main/scala/cromwell/server/WorkflowManagerSystem.scala

@@ -9,5 +9,13 @@ trait WorkflowManagerSystem {
   implicit val actorSystem = ActorSystem(systemName)
   lazy val workflowManagerActor = actorSystem.actorOf(WorkflowManagerActor.props(dataAccess))
 
-  def dataAccess: DataAccess
-}
+  // Lazily created as the primary consumer is the workflowManagerActor.
+  private lazy val dataAccess: DataAccess = DataAccess()
+
+  /**
+   * Should be called after the system is no longer in use.
+   *
+   * @return Non-blocking future that will eventually shut down this instance or return an error.
+   */
+  def shutdown() = dataAccess.shutdown()
+}

src/test/scala/cromwell/MainSpec.scala

@@ -3,17 +3,16 @@ package cromwell
 import akka.actor.ActorSystem
 import akka.testkit.EventFilter
 import com.typesafe.config.ConfigFactory
-import cromwell.engine.db.DummyDataAccess
-import cromwell.engine.workflow.WorkflowManagerActor
 import cromwell.server.WorkflowManagerSystem
 import cromwell.util.FileUtil
 import cromwell.util.SampleWdl.ThreeStep
 import org.scalatest.{FlatSpec, Matchers}
 
+import scala.concurrent.Await
+import scala.concurrent.duration.Duration
+
 class TestWorkflowManagerSystem extends WorkflowManagerSystem {
-  override def dataAccess = DummyDataAccess()
   override implicit val actorSystem = ActorSystem(systemName, ConfigFactory.parseString(CromwellTestkitSpec.ConfigText))
-  override lazy val workflowManagerActor = actorSystem.actorOf(WorkflowManagerActor.props(dataAccess))
 }
 
 class MainSpec extends FlatSpec with Matchers {
@@ -102,12 +101,12 @@ class MainSpec extends FlatSpec with Matchers {
   }
 
   it should "run" in {
-    val stream = new java.io.ByteArrayOutputStream()
     val workflowManagerSystem = new TestWorkflowManagerSystem
     implicit val system = workflowManagerSystem.actorSystem
     EventFilter.info(pattern = s"workflow finished", occurrences = 1).intercept {
       Main.run(Array(wdlFilePathAndWriter._1.toAbsolutePath.toString, inputsJsonPathAndWriter._1.toAbsolutePath.toString), workflowManagerSystem)
     }
+    Await.result(workflowManagerSystem.shutdown(), Duration.Inf)
   }
 
   it should "print usage" in {