forked from joelplourde3/clin-pipelines
-
Notifications
You must be signed in to change notification settings - Fork 0
/
MinioServer.scala
96 lines (79 loc) · 3.67 KB
/
MinioServer.scala
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
package bio.ferlab.clin.etl.fhir.testutils
import bio.ferlab.clin.etl.conf.AWSConf
import bio.ferlab.clin.etl.fhir.testutils.FhirTestUtils.getClass
import bio.ferlab.clin.etl.fhir.testutils.containers.MinioContainer
import bio.ferlab.clin.etl.s3.S3Utils
import org.scalatest.{BeforeAndAfterAll, TestSuite}
import org.slf4j.{Logger, LoggerFactory}
import software.amazon.awssdk.core.sync.RequestBody
import software.amazon.awssdk.services.s3.S3Client
import software.amazon.awssdk.services.s3.model.{CreateBucketRequest, DeleteObjectRequest, ListObjectsRequest, PutObjectRequest}
import java.io.File
import java.nio.file.Path
import scala.collection.JavaConverters._
import scala.util.Random
trait MinioServer {
private val minioPort = MinioContainer.startIfNotRunning()
protected val minioEndpoint = s"http://localhost:${minioPort}"
implicit val s3: S3Client = S3Utils.buildS3Client(AWSConf(MinioContainer.accessKey, MinioContainer.secretKey, minioEndpoint, pathStyleAccess = true, "", "", ""))
val LOGGER: Logger = LoggerFactory.getLogger(getClass)
val inputBucket = s"clin-import"
val outputBucket = s"clin-repository"
createBuckets()
private def createBuckets(): Unit = {
val alreadyExistingBuckets = s3.listBuckets().buckets().asScala.collect { case b if b.name() == inputBucket || b.name() == outputBucket => b.name() }
val bucketsToCreate = Seq(inputBucket, outputBucket).diff(alreadyExistingBuckets)
bucketsToCreate.foreach { b =>
val buketRequest = CreateBucketRequest.builder().bucket(b).build()
s3.createBucket(buketRequest)
}
}
def withS3Objects[T](block: (String, String) => T): Unit = {
val inputPrefix = s"run_${Random.nextInt(10000)}"
LOGGER.info(s"Use input prefix $inputPrefix : $minioEndpoint/minio/$inputBucket/$inputPrefix")
val outputPrefix = s"files_${Random.nextInt(10000)}"
LOGGER.info(s"Use output prefix $outputPrefix : : $minioEndpoint/minio/$outputBucket/$outputPrefix")
try {
block(inputPrefix, outputPrefix)
} finally {
deleteRecursively(inputBucket, inputPrefix)
deleteRecursively(outputBucket, outputPrefix)
}
}
def list(bucket: String, prefix: String): Seq[String] = {
val lsRequest = ListObjectsRequest.builder().bucket(bucket).prefix(prefix).build()
s3.listObjects(lsRequest).contents().asScala.map(_.key())
}
private def deleteRecursively(bucket: String, prefix: String): Unit = {
val lsRequest = ListObjectsRequest.builder().bucket(bucket).prefix(prefix).build()
s3.listObjects(lsRequest).contents().asScala.foreach { o =>
val del = DeleteObjectRequest.builder().bucket(bucket).key(prefix).build()
s3.deleteObject(del)
}
}
def ls(file: File): List[File] = {
file.listFiles.filter(_.isFile).toList
}
def transferFromResources(prefix: String, resource: String, bucket: String = inputBucket): Unit = {
val files = ls(new File(getClass.getResource(s"/$resource").toURI))
files.foreach { f =>
val put = PutObjectRequest.builder().bucket(bucket).key(s"$prefix/${f.getName}").build()
s3.putObject(put, RequestBody.fromFile(f))
}
}
def copyNFile(prefix: String, resource: String, times: Int, bucket: String = inputBucket): Unit = {
val file = new File(getClass.getResource(s"/$resource").toURI)
1.to(times).map { i =>
val filename = s"${file.getName}_$i"
val put = PutObjectRequest.builder().bucket(bucket).key(s"$prefix/$filename").build()
s3.putObject(put, RequestBody.fromFile(file))
}
}
}
trait MinioServerSuite extends MinioServer with TestSuite with BeforeAndAfterAll {
}
object StartMinioServer extends App with MinioServer {
LOGGER.info(s"Minio is started : $minioEndpoint")
while (true) {
}
}