<commit_before>package dit4c.machineshop.docker

import java.util.Calendar
import java.util.TimeZone
import java.util.concurrent.Executors

import scala.collection.JavaConversions.asScalaBuffer
import scala.concurrent.ExecutionContext
import scala.concurrent.Future
import scala.concurrent.duration.Duration

import com.github.dockerjava.api.model.Link
import com.github.dockerjava.api.model.RestartPolicy
import com.github.dockerjava.api.model._
import com.github.dockerjava.core.DockerClientBuilder
import com.github.dockerjava.core.DockerClientConfig
import com.github.dockerjava.core.command.PullImageResultCallback

import akka.http.scaladsl.model.Uri
import akka.stream.io.InputStreamSource
import akka.stream.scaladsl.Source
import akka.util.ByteString
import dit4c.machineshop.docker.models.ContainerLink
import dit4c.machineshop.docker.models.ContainerStatus
import dit4c.machineshop.docker.models.DockerContainer
import dit4c.machineshop.docker.models.DockerContainers
import dit4c.machineshop.docker.models.DockerImage
import dit4c.machineshop.docker.models.DockerImages
import dit4c.machineshop.docker.utils.DiskBasedChunker

class DockerClientImpl(
    val baseUrl: Uri,
    val newContainerLinks: Seq[ContainerLink] = Nil) extends DockerClient {

  // Uncapped ExecutionContext, so we can do sync network requests
  implicit val ec = ExecutionContext.fromExecutor(Executors.newCachedThreadPool)

  val docker: com.github.dockerjava.api.DockerClient = {
    val config =
      DockerClientConfig.createDefaultConfigBuilder
        .withUri(baseUrl.toString)
        .build
    DockerClientBuilder.getInstance(config).build
  }

  override val images = ImagesImpl
  override val containers = ContainersImpl

  case class ImageImpl(
      val id: String,
      val names: Set[String],
      val created: Calendar) extends DockerImage

  object ImagesImpl extends DockerImages {

    def list = Future({
      docker.listImagesCmd.exec.toIterable.map { img =>
        ImageImpl(
            img.getId,
            img.getRepoTags.toSet.filter(_ != "<none>:<none>"),
            {
              val c = Calendar.getInstance(TimeZone.getTimeZone("GMT"))
              c.setTimeInMillis(img.getCreated * 1000)
              c
            })
      }.filter(!_.names.isEmpty).toSeq
    })(ec)

    def pull(imageName: String, tagName: String) = Future({
      docker.pullImageCmd(imageName).withTag(tagName)
        .exec(new PullImageResultCallback()).awaitSuccess
    })(ec)

  }



  class ContainerImpl(val id: String, val name: String, val status: ContainerStatus) extends DockerContainer {

    override def refresh = Future({
      val c = docker.inspectContainerCmd(id).exec
      val status =
        if (c.getState.isRunning) ContainerStatus.Running
        else ContainerStatus.Stopped
      new ContainerImpl(id, name, status)
    })(ec)

    override def start = Future({
      docker.startContainerCmd(id).exec
    })(ec).flatMap(_ => this.refresh)

    override def stop(timeout: Duration) = Future({
      docker.stopContainerCmd(id)
        .withTimeout(Math.max(0, timeout.toSeconds.toInt)).exec
    })(ec).flatMap(_ => this.refresh)

    override def export: Future[Source[ByteString, Future[Long]]] = Future({
      docker.commitCmd(id).exec
    })(ec).map { imageId =>
      lazy val fRemoveImage = Future(docker.removeImageCmd(imageId).exec)(ec)
      InputStreamSource(() =>docker.saveImageCmd(imageId).exec)
        .transform(() => new DiskBasedChunker(65536))
        .map(v => { fRemoveImage; v }) // Remove image onces stream starts
    }

    override def delete = Future({
      docker.removeContainerCmd(id).withRemoveVolumes(true).exec
      () // Need to return Unit, not Void
    })(ec)

  }<commit_msg>Increasing chunk size<commit_after>package dit4c.machineshop.docker

import java.util.Calendar
import java.util.TimeZone
import java.util.concurrent.Executors

import scala.collection.JavaConversions.asScalaBuffer
import scala.concurrent.ExecutionContext
import scala.concurrent.Future
import scala.concurrent.duration.Duration

import com.github.dockerjava.api.model.Link
import com.github.dockerjava.api.model.RestartPolicy
import com.github.dockerjava.api.model._
import com.github.dockerjava.core.DockerClientBuilder
import com.github.dockerjava.core.DockerClientConfig
import com.github.dockerjava.core.command.PullImageResultCallback

import akka.http.scaladsl.model.Uri
import akka.stream.io.InputStreamSource
import akka.stream.scaladsl.Source
import akka.util.ByteString
import dit4c.machineshop.docker.models.ContainerLink
import dit4c.machineshop.docker.models.ContainerStatus
import dit4c.machineshop.docker.models.DockerContainer
import dit4c.machineshop.docker.models.DockerContainers
import dit4c.machineshop.docker.models.DockerImage
import dit4c.machineshop.docker.models.DockerImages
import dit4c.machineshop.docker.utils.DiskBasedChunker

class DockerClientImpl(
    val baseUrl: Uri,
    val newContainerLinks: Seq[ContainerLink] = Nil) extends DockerClient {

  // Uncapped ExecutionContext, so we can do sync network requests
  implicit val ec = ExecutionContext.fromExecutor(Executors.newCachedThreadPool)

  val docker: com.github.dockerjava.api.DockerClient = {
    val config =
      DockerClientConfig.createDefaultConfigBuilder
        .withUri(baseUrl.toString)
        .build
    DockerClientBuilder.getInstance(config).build
  }

  override val images = ImagesImpl
  override val containers = ContainersImpl

  case class ImageImpl(
      val id: String,
      val names: Set[String],
      val created: Calendar) extends DockerImage

  object ImagesImpl extends DockerImages {

    def list = Future({
      docker.listImagesCmd.exec.toIterable.map { img =>
        ImageImpl(
            img.getId,
            img.getRepoTags.toSet.filter(_ != "<none>:<none>"),
            {
              val c = Calendar.getInstance(TimeZone.getTimeZone("GMT"))
              c.setTimeInMillis(img.getCreated * 1000)
              c
            })
      }.filter(!_.names.isEmpty).toSeq
    })(ec)

    def pull(imageName: String, tagName: String) = Future({
      docker.pullImageCmd(imageName).withTag(tagName)
        .exec(new PullImageResultCallback()).awaitSuccess
    })(ec)

  }



  class ContainerImpl(val id: String, val name: String, val status: ContainerStatus) extends DockerContainer {

    override def refresh = Future({
      val c = docker.inspectContainerCmd(id).exec
      val status =
        if (c.getState.isRunning) ContainerStatus.Running
        else ContainerStatus.Stopped
      new ContainerImpl(id, name, status)
    })(ec)

    override def start = Future({
      docker.startContainerCmd(id).exec
    })(ec).flatMap(_ => this.refresh)

    override def stop(timeout: Duration) = Future({
      docker.stopContainerCmd(id)
        .withTimeout(Math.max(0, timeout.toSeconds.toInt)).exec
    })(ec).flatMap(_ => this.refresh)

    override def export: Future[Source[ByteString, Future[Long]]] = Future({
      docker.commitCmd(id).exec
    })(ec).map { imageId =>
      lazy val fRemoveImage = Future(docker.removeImageCmd(imageId).exec)(ec)
      InputStreamSource(() =>docker.saveImageCmd(imageId).exec)
        .transform(() => new DiskBasedChunker(250000))
        .map(v => { fRemoveImage; v }) // Remove image onces stream starts
    }

    override def delete = Future({
      docker.removeContainerCmd(id).withRemoveVolumes(true).exec
      () // Need to return Unit, not Void
    })(ec)

  }
{'fraction_non_alphanumeric': 0.10116108367809956, 'fraction_numerical': 0.004937942079273989, 'mean_word_length': 3.6575512740832816, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5328708', 'n_tokens_mistral': 2292, 'n_tokens_neox': 2144, 'n_words': 446}