S3/MinIO с Java/Scala: сохранение байтовых буферов фрагментов файлов в хранилище объектов

Я понял, что я делал неправильно (при использовании клиента S3). Я не накапливал байты внутри моего upload.handler. Мне нужно накапливать байты до тех пор, пока размер буфера не станет достаточно большим для загрузки части, а не загружать каждый раз, когда я получаю несколько байтов.

Поскольку ни клиент Amazon S3, ни клиент MinIO не делали того, что мне нужно, я решил изучить, как на самом деле был реализован putObject(), и сделать свой собственный. Это то, что я придумал.

Эта реализация специфична для Vert.X, однако ее можно легко обобщить для работы со встроенными java.io InputStreams через while цикл и с использованием пары Piped- потоков.

Эта реализация также специфична для MinIO, но ее можно легко адаптировать для использования клиента S3, поскольку по большей части эти два API одинаковы.

В этом примере Buffer в основном представляет собой контейнер вокруг ByteArray, и я не делаю здесь ничего особенного. Я заменил его массивом байтов, чтобы гарантировать, что он все еще будет работать, и это сработало.

package server

import com.google.common.collect.HashMultimap
import io.minio.MinioClient
import io.minio.messages.Part
import io.vertx.core.buffer.Buffer
import io.vertx.core.streams.ReadStream

import scala.collection.mutable.ListBuffer

class CustomMinioClient(client: MinioClient) extends MinioClient(client) {
  def putReadStream(bucket: String = "my-bucket",
                    objectName: String,
                    region: String = "us-east-1",
                    data: ReadStream[Buffer],
                    objectSize: Long,
                    contentType: String = "application/octet-stream"
                   ) = {
    val headers: HashMultimap[String, String] = HashMultimap.create()
    headers.put("Content-Type", contentType)
    var uploadId: String = null

    try {
      val parts = new ListBuffer[Part]()
      val createResponse = createMultipartUpload(bucket, region, objectName, headers, null)
      uploadId = createResponse.result.uploadId()

      var partNumber = 1
      var uploadedSize = 0

      // an array to use to accumulate bytes from the incoming stream until we have enough to make a `uploadPart` request
      var partBuffer = Buffer.buffer()

      // S3's minimum part size is 5mb, excepting the last part
      // you should probably implement your own logic for determining how big
      // to make each part based off the total object size to avoid unnecessary calls to S3 to upload small parts.
      val minPartSize = 5 * 1024 * 1024

      data.handler { buffer =>

        partBuffer.appendBuffer(buffer)

        val availableSize = objectSize - uploadedSize - partBuffer.length

        val isMinPartSize = partBuffer.length >= minPartSize
        val isLastPart = uploadedSize + partBuffer.length == objectSize

        if (isMinPartSize || isLastPart) {

          val partResponse = uploadPart(
            bucket,
            region,
            objectName,
            partBuffer.getBytes,
            partBuffer.length,
            uploadId,
            partNumber,
            null,
            null
          )

          parts.addOne(new Part(partNumber, partResponse.etag))
          uploadedSize += partBuffer.length
          partNumber += 1

          // empty the part buffer since we have already uploaded it
          partBuffer = Buffer.buffer()
        }
      }


      data.endHandler { _ =>
        completeMultipartUpload(bucket, region, objectName, uploadId, parts.toArray, null, null)
      }

      data.exceptionHandler { exception =>
        // should also probably abort the upload here
        println("Handler caught exception in custom putObject: " + exception)
      }
    } catch {
      // and abort it here as well...
      case e: Exception =>
        println("Exception thrown in custom `putObject`: " + e)
        abortMultipartUpload(
          bucket,
          region,
          objectName,
          uploadId,
          null,
          null
        )
    }
  }
}

Все это можно использовать довольно легко.

Сначала настройте клиент:

  private val _minioClient = MinioClient.builder()
    .endpoint("http://localhost:9000")
    .credentials("my-username", "my-password")
    .build()

  private val myClient = new CustomMinioClient(_minioClient)

Затем, когда вы получаете запрос на загрузку:

      context.request.uploadHandler { upload =>
        myClient.putReadStream(objectName = upload.filename(), data = upload, objectSize = myFileSize)
        context.response().setStatusCode(200).end("done")
      }

Единственная загвоздка в этой реализации заключается в том, что вам нужно заранее знать размеры файлов для запроса.

Однако это можно легко решить так, как это сделал я, особенно если вы используете веб-интерфейс.

• Прежде чем пытаться загрузить файлы, отправьте запрос на сервер, содержащий сопоставление имени файла с размером файла.
• Этот предварительный запрос должен генерировать уникальный идентификатор для загрузки.
• Сервер может сохранить группу имя файла-›размер файла, используя идентификатор загрузки в качестве индекса. - Сервер отправляет идентификатор загрузки обратно клиенту.
• Клиент отправляет составной запрос на загрузку, используя идентификатор загрузки.
• Сервер извлекает список файлов и их размеров и использует его для вызова .putReadStream()