feat: add support for memcached

Author: j5ik2o

build.sbt

@@ -40,11 +40,15 @@ lazy val baseSettings = Seq(
   resolvers ++= Seq(
     Resolver.sonatypeRepo("snapshots"),
     Resolver.sonatypeRepo("releases"),
-    "Seasar Repository" at "https://maven.seasar.org/maven2/"
+    "Seasar Repository" at "https://maven.seasar.org/maven2/",
+    "Spy Repository" at "https://files.couchbase.com/maven2/"
   ),
   libraryDependencies ++= Seq(
     scalatest.scalatest % Test
   ),
+  dependencyOverrides ++= Seq(
+    "com.fasterxml.jackson.core" % "jackson-databind" % "2.11.0"
+  ),
   ThisBuild / scalafixScalaBinaryVersion := CrossVersion.binaryScalaVersion(scalaVersion.value),
   semanticdbEnabled := true,
   semanticdbVersion := scalafixSemanticdb.revision,
@@ -193,6 +197,17 @@ val `docker-controller-scala-redis` = (project in file("docker-controller-scala-
     )
   ).dependsOn(`docker-controller-scala-core`, `docker-controller-scala-scalatest` % Test)
 
+val `docker-controller-scala-memcached` = (project in file("docker-controller-scala-memcached"))
+  .settings(baseSettings)
+  .settings(
+    name := "docker-controller-scala-memcached",
+    libraryDependencies ++= Seq(
+      scalatest.scalatest       % Test,
+      logback.classic           % Test,
+      (twitter.finagleMemcached % Test).cross(CrossVersion.for3Use2_13)
+    )
+  ).dependsOn(`docker-controller-scala-core`, `docker-controller-scala-scalatest` % Test)
+
 val `docker-controller-scala-elasticmq` = (project in file("docker-controller-scala-elasticmq"))
   .settings(baseSettings)
   .settings(
@@ -233,11 +248,13 @@ val `docker-controller-scala-root` = (project in file("."))
   .aggregate(
     `docker-controller-scala-core`,
     `docker-controller-scala-scalatest`,
+    `docker-controller-scala-flyway`,
     // for RDBMS
     `docker-controller-scala-mysql`,
     `docker-controller-scala-postgresql`,
     // for NoSQL
     `docker-controller-scala-redis`,
+    `docker-controller-scala-memcached`,
     `docker-controller-scala-elasticsearch`,
     `docker-controller-scala-kafka`,
     `docker-controller-scala-zookeeper`,

docker-controller-scala-core/src/main/scala/com/github/j5ik2o/dockerController/DockerController.scala

@@ -58,7 +58,7 @@ trait DockerController {
   def existsImage(p: Image => Boolean): Boolean
   def pullImageIfNotExists(f: PullImageCmd => PullImageCmd = identity): Unit
   def pullImage(f: PullImageCmd => PullImageCmd = identity): Unit
-  def awaitCondition(duration: Duration)(predicate: Frame => Boolean): Unit
+  def awaitCondition(duration: Duration)(predicate: Option[Frame] => Boolean): Unit
 }
 
 object DockerController {
@@ -247,7 +247,7 @@ private[dockerController] class DockerControllerImpl(
     logger.debug("stopContainer --- finish")
   }
 
-  override def awaitCondition(duration: Duration)(predicate: Frame => Boolean): Unit = {
+  override def awaitCondition(duration: Duration)(predicate: Option[Frame] => Boolean): Unit = {
     logger.debug("awaitCompletion --- start")
     val frameQueue: LinkedBlockingQueue[Frame] = new LinkedBlockingQueue[Frame]()
 
@@ -266,11 +266,11 @@ private[dockerController] class DockerControllerImpl(
         def loop(): Unit = {
           if (
             !terminate && {
-              val frame = frameQueue.poll(outputFrameInterval.toMillis, TimeUnit.MILLISECONDS)
-              if (frame != null) {
+              val frameOpt = Option(frameQueue.poll(outputFrameInterval.toMillis, TimeUnit.MILLISECONDS))
+              frameOpt.foreach { frame =>
                 logger.debug(frame.toString)
-                !predicate(frame)
-              } else true
+              }
+              !predicate(frameOpt)
             }
           ) {
             loop()

docker-controller-scala-core/src/main/scala/com/github/j5ik2o/dockerController/WaitPredicates.scala

@@ -10,56 +10,73 @@ import scala.util.matching.Regex
 
 object WaitPredicates {
 
-  type WaitPredicate = Frame => Boolean
+  type WaitPredicate = Option[Frame] => Boolean
 
   protected val logger: Logger = LoggerFactory.getLogger(getClass)
 
+  def forDebug(
+      awaitDurationOpt: Option[FiniteDuration] = Some(500.milliseconds)
+  ): WaitPredicate = { frameOpt =>
+    frameOpt.exists { frame =>
+      val line = new String(frame.getPayload).stripLineEnd
+      logger.debug(s"forDebug: line = $line")
+      awaitDurationOpt.foreach { awaitDuration => Thread.sleep(awaitDuration.toMillis) }
+      false
+    }
+  }
+
   def forLogMessageExactly(
       text: String,
       awaitDurationOpt: Option[FiniteDuration] = Some(500.milliseconds)
-  ): WaitPredicate = { frame =>
-    val line   = new String(frame.getPayload).stripLineEnd
-    val result = line == text
-    if (result) {
-      logger.debug(s"forLogMessageExactly: result = $result, line = $line")
-      awaitDurationOpt.foreach { awaitDuration => Thread.sleep(awaitDuration.toMillis) }
+  ): WaitPredicate = { frameOpt =>
+    frameOpt.exists { frame =>
+      val line   = new String(frame.getPayload).stripLineEnd
+      val result = line == text
+      if (result) {
+        logger.debug(s"forLogMessageExactly: result = $result, line = $line")
+        awaitDurationOpt.foreach { awaitDuration => Thread.sleep(awaitDuration.toMillis) }
+      }
+      result
     }
-    result
   }
 
   def forLogMessageContained(
       text: String,
       awaitDurationOpt: Option[FiniteDuration] = Some(500.milliseconds)
-  ): WaitPredicate = { frame =>
-    val line   = new String(frame.getPayload).stripLineEnd
-    val result = line.contains(text)
-    if (result) {
-      logger.debug(s"forLogMessageContained: result = $result, line = $line")
-      awaitDurationOpt.foreach { awaitDuration => Thread.sleep(awaitDuration.toMillis) }
+  ): WaitPredicate = { frameOpt =>
+    frameOpt.exists { frame =>
+      val line   = new String(frame.getPayload).stripLineEnd
+      val result = line.contains(text)
+      if (result) {
+        logger.debug(s"forLogMessageContained: result = $result, line = $line")
+        awaitDurationOpt.foreach { awaitDuration => Thread.sleep(awaitDuration.toMillis) }
+      }
+      result
     }
-    result
   }
 
   def forLogMessageByRegex(
       regex: Regex,
       awaitDurationOpt: Option[FiniteDuration] = Some(500.milliseconds)
-  ): WaitPredicate = { frame =>
-    val line   = new String(frame.getPayload).stripLineEnd
-    val result = regex.findFirstIn(line).isDefined
-    if (result) {
-      logger.debug(s"forLogMessageByRegex: result = $result, line = $line")
-      awaitDurationOpt.foreach { awaitDuration => Thread.sleep(awaitDuration.toMillis) }
+  ): WaitPredicate = { frameOpt =>
+    frameOpt.exists { frame =>
+      val line   = new String(frame.getPayload).stripLineEnd
+      val result = regex.findFirstIn(line).isDefined
+      if (result) {
+        logger.debug(s"forLogMessageByRegex: result = $result, line = $line")
+        awaitDurationOpt.foreach { awaitDuration => Thread.sleep(awaitDuration.toMillis) }
+      }
+      result
     }
-    result
   }
 
   def forListeningHostTcpPort(
       host: String,
       hostPort: Int,
       connectionTimeout: FiniteDuration = 500.milliseconds,
       awaitDurationOpt: Option[FiniteDuration] = Some(500.milliseconds)
-  ): WaitPredicate = { f: Frame =>
-    val line      = new String(f.getPayload)
+  ): WaitPredicate = { frameOpt =>
+    val line      = frameOpt.map(frame => new String(frame.getPayload)).getOrElse("")
     val s: Socket = new Socket()
     try {
       s.connect(new InetSocketAddress(host, hostPort), connectionTimeout.toMillis.toInt)
@@ -82,13 +99,13 @@ object WaitPredicates {
       host: String,
       hostPort: Int,
       awaitDurationOpt: Option[FiniteDuration] = Some(500.milliseconds)
-  ): WaitPredicate = { _: Frame => forListeningHttp(host, hostPort, awaitDurationOpt).isDefined }
+  ): WaitPredicate = { _ => forListeningHttp(host, hostPort, awaitDurationOpt).isDefined }
 
   def forListeningHttpPortWithPredicate(
       host: String,
       hostPort: Int,
       awaitDurationOpt: Option[FiniteDuration] = Some(500.milliseconds)
-  )(p: HttpURLConnection => Boolean): WaitPredicate = { _: Frame =>
+  )(p: HttpURLConnection => Boolean): WaitPredicate = { _ =>
     forListeningHttp(host, hostPort, awaitDurationOpt).exists(p)
   }
 

docker-controller-scala-kafka/src/main/scala/com/github/j5ik2o/dockerController/kafka/KafkaController.scala

@@ -104,7 +104,7 @@ class KafkaController(
     dockerClient.removeNetworkCmd(networkId).exec()
   }
 
-  override def awaitCondition(duration: Duration)(predicate: Frame => Boolean): Unit = {
+  override def awaitCondition(duration: Duration)(predicate: Option[Frame] => Boolean): Unit = {
     zooKeeperController.awaitCondition(duration)(zooKeeperWaitPredicate)
     super.awaitCondition(duration)(predicate)
   }

docker-controller-scala-memcached/src/main/scala/com/github/j5ik2o/dockerController/memcached/MemcachedController.scala

@@ -0,0 +1,58 @@
+package com.github.j5ik2o.dockerController.memcached
+
+import com.github.dockerjava.api.DockerClient
+import com.github.dockerjava.api.command.CreateContainerCmd
+import com.github.dockerjava.api.model.{ ExposedPort, Ports }
+import com.github.dockerjava.api.model.HostConfig.newHostConfig
+import com.github.j5ik2o.dockerController.DockerControllerImpl
+import com.github.j5ik2o.dockerController.memcached.MemcachedController._
+
+import scala.concurrent.duration._
+
+object MemcachedController {
+  final val DefaultImageName: String        = "memcached"
+  final val DefaultImageTag: Option[String] = Some("1.6")
+  final val DefaultContainerPort: Int       = 11211
+
+  def apply(
+      dockerClient: DockerClient,
+      outputFrameInterval: FiniteDuration = 500.millis,
+      imageName: String = DefaultImageName,
+      imageTag: Option[String] = DefaultImageTag,
+      envVars: Map[String, String] = Map.empty
+  )(
+      hostPort: Int,
+      prometheusEnabled: Boolean = false
+  ): MemcachedController = new MemcachedController(dockerClient, outputFrameInterval, imageName, imageTag, envVars)(
+    hostPort,
+    prometheusEnabled
+  )
+}
+
+class MemcachedController(
+    dockerClient: DockerClient,
+    outputFrameInterval: FiniteDuration = 500.millis,
+    imageName: String = DefaultImageName,
+    imageTag: Option[String] = DefaultImageTag,
+    envVars: Map[String, String] = Map.empty
+)(
+    hostPort: Int,
+    prometheusEnabled: Boolean = false
+) extends DockerControllerImpl(dockerClient, outputFrameInterval)(imageName, imageTag) {
+
+  private val environmentVariables = Map(
+    "MEMCACHED_PROMETHEUS_ENABLED" -> prometheusEnabled.toString
+  ) ++
+    envVars
+
+  override protected def newCreateContainerCmd(): CreateContainerCmd = {
+    val containerPort = ExposedPort.tcp(DefaultContainerPort)
+    val portBinding   = new Ports
+    portBinding.bind(containerPort, Ports.Binding.bindPort(hostPort))
+    super
+      .newCreateContainerCmd()
+      .withEnv(environmentVariables.map { case (k, v) => s"$k=$v" }.toArray: _*)
+      .withExposedPorts(containerPort)
+      .withHostConfig(newHostConfig().withPortBindings(portBinding))
+  }
+}

docker-controller-scala-memcached/src/test/scala/com/github/j5ik2o/dockerController/memcached/MemcachedControllerSpec.scala

@@ -0,0 +1,50 @@
+package com.github.j5ik2o.dockerController.memcached
+
+import com.github.j5ik2o.dockerController.{
+  DockerController,
+  DockerControllerSpecSupport,
+  RandomPortUtil,
+  WaitPredicates
+}
+import com.twitter.finagle.Memcached
+import com.twitter.io.Buf
+import org.scalatest.concurrent.ScalaFutures
+import org.scalatest.freespec.AnyFreeSpec
+
+import scala.concurrent.duration._
+import scala.jdk.FutureConverters.CompletionStageOps
+
+class MemcachedControllerSpec extends AnyFreeSpec with DockerControllerSpecSupport with ScalaFutures {
+  val testTimeFactor: Int = sys.env.getOrElse("TEST_TIME_FACTOR", "1").toInt
+  logger.debug(s"testTimeFactor = $testTimeFactor")
+
+  val hostPort: Int                   = RandomPortUtil.temporaryServerPort()
+  val controller: MemcachedController = MemcachedController(dockerClient)(hostPort)
+
+  override protected val dockerControllers: Vector[DockerController] = Vector(controller)
+
+  override protected val waitPredicatesSettings: Map[DockerController, WaitPredicateSetting] = Map(
+    controller -> WaitPredicateSetting(
+      Duration.Inf,
+      WaitPredicates.forListeningHostTcpPort(
+        dockerHost,
+        hostPort,
+        (1 * testTimeFactor).seconds,
+        Some((5 * testTimeFactor).seconds)
+      )
+    )
+  )
+
+  "MemcachedController" - {
+    "run" in {
+      val client = Memcached.client.newRichClient(s"$dockerHost:$hostPort")
+      val str    = "a"
+      val buf    = Buf.Utf8(str)
+      val result = for {
+        _ <- client.set("1", buf)
+        r <- client.get("1")
+      } yield r
+      assert(result.toCompletableFuture.asScala.futureValue.get == buf)
+    }
+  }
+}

project/Dependencies.scala

@@ -107,4 +107,8 @@ object Dependencies {
     val redisClient = "net.debasishg" %% "redisclient" % "3.30"
   }
 
+  object twitter {
+    val finagleMemcached = "com.twitter" %% "finagle-memcached" % "21.6.0"
+  }
+
 }