Uploading a file with progress in Kotlin #Retrofit https://vonlisboa.medium.com/uploading-a-file-with-progress-in-kotlin-6afbfd80fb1c

AttachmentRemoteApi.kt

// https://vonlisboa.medium.com/uploading-a-file-with-progress-in-kotlin-6afbfd80fb1c
package com.example.retrofit.upload

import io.reactivex.rxjava3.core.Single
import okhttp3.MultipartBody
import okhttp3.RequestBody
import retrofit2.http.Multipart
import retrofit2.http.POST
import retrofit2.http.Part


interface AttachmentRemoteApi {
    @Multipart
    @POST("file_uploader")
    fun attachFile(
        @Part("name") filename: RequestBody,
        @Part("type") mimeType: RequestBody,
        @Part("size") fileSize: RequestBody,
        @Part files: MultipartBody.Part
    ): Single<AttachmentUploadedRemoteDto>
}

AttachmentUploadedRemoteDto.kt

package com.example.retrofit.upload

import com.google.gson.annotations.SerializedName
import okhttp3.MultipartBody

data class AttachmentUploadedRemoteDto(
        @SerializedName("filename") var filename: String,
        @SerializedName( "mimeType") var mimeType: String,
        @SerializedName( "fileSize") var fileSize: String,
        @SerializedName( "filePart") var filePart: MultipartBody.Part,
)

AttachmentUploader.kt

package com.example.retrofit.upload

import com.google.gson.GsonBuilder
import io.reactivex.rxjava3.core.Observable
import io.reactivex.rxjava3.core.Single
import io.reactivex.rxjava3.subjects.PublishSubject
import okhttp3.MediaType.Companion.toMediaType
import okhttp3.MultipartBody
import okhttp3.RequestBody
import okhttp3.RequestBody.Companion.asRequestBody
import okhttp3.RequestBody.Companion.toRequestBody
import retrofit2.Retrofit
import retrofit2.adapter.rxjava3.RxJava3CallAdapterFactory
import retrofit2.converter.gson.GsonConverterFactory
import java.io.File


class AttachmentUploader {
    private var remoteApi: AttachmentRemoteApi
    init {
        val gson = GsonBuilder().setLenient().create()
        val retrofit = Retrofit.Builder()
                .baseUrl("http://yoururl.com")
                .addConverterFactory(GsonConverterFactory.create(gson))
                .addCallAdapterFactory(RxJava3CallAdapterFactory.create())
                .build()

        remoteApi = retrofit.create(AttachmentRemoteApi::class.java)
    }

    private fun String.toPlainTextBody() = toRequestBody("text/plain".toMediaType())

    private fun createUploadRequestBody(
            file: File,
            mimeType: String,
            progressEmitter: PublishSubject<Double>
    ): RequestBody {
        val fileRequestBody = file.asRequestBody(mimeType.toMediaType())
        return CountingRequestBody(fileRequestBody) { bytesWritten, contentLength ->
            val progress = 1.0 * bytesWritten / contentLength
            if (progress >= 1.0) {
                progressEmitter.onComplete()
            }else{
                progressEmitter.onNext(progress)
            }
        }
    }

    private fun createUploadRequest(
            filename: String,
            file: File,
            mimeType: String,
            progressEmitter: PublishSubject<Double>
    ): Single<AttachmentUploadedRemoteDto> {
        val requestBody = createUploadRequestBody(file, mimeType, progressEmitter)
        return remoteApi.attachFile(
                filename = filename.toPlainTextBody(),
                mimeType = mimeType.toPlainTextBody(),
                fileSize = file.length().toString().toPlainTextBody(),
                files = MultipartBody.Part.createFormData(
                        "files",
                        filename,
                        requestBody
                )
        )
    }

    fun uploadAttachment(
            filename: String, file: File, mimeType: String
    ): Observable<AttachmentUploadRemoteResult> {
        val progressEmitter = PublishSubject.create<Double>()
        val uploadRequest = createUploadRequest(
                filename, file, mimeType, progressEmitter
        )

        val uploadResult = uploadRequest
                .map<AttachmentUploadRemoteResult> {
                    CountingRequestResult.Completed(it) // it.result ???
                }
                .toObservable()

        val progressResult = progressEmitter
                .map<AttachmentUploadRemoteResult> {
                    CountingRequestResult.Progress(it)
                }

        return progressResult.mergeWith(uploadResult)

    }
}



typealias AttachmentUploadRemoteResult = CountingRequestResult<AttachmentUploadedRemoteDto>

build.gradle

implementation "com.squareup.retrofit2:converter-moshi:2.4.0"
implementation 'com.squareup.retrofit2:converter-gson:2.1.0'
implementation 'com.squareup.okio:okio:2.10.0'
implementation 'com.squareup.okhttp3:okhttp:4.9.0'
implementation 'io.reactivex.rxjava3:rxkotlin:3.0.1'
implementation "io.reactivex.rxjava3:rxandroid:3.0.0"
implementation 'com.squareup.retrofit2:adapter-rxjava3:2.9.0'

CountingRequestBody.kt

package com.example.retrofit.upload

import okhttp3.RequestBody
import okio.BufferedSink
import okio.buffer
import java.io.IOException

class CountingRequestBody(
        private val requestBody: RequestBody,
        private val onProgressUpdate: CountingRequestListener
) : RequestBody() {
    override fun contentType() = requestBody.contentType()

    @Throws(IOException::class)
    override fun contentLength() = requestBody.contentLength()

    @Throws(IOException::class)
    override fun writeTo(sink: BufferedSink) {
        val countingSink = CountingSink(sink, this, onProgressUpdate)
        val bufferedSink = countingSink.buffer()
        requestBody.writeTo(bufferedSink)
        bufferedSink.flush()
    }
}

CountingRequestResult.kt

package com.example.retrofit.upload

sealed class CountingRequestResult<ResultT> {
    data class Progress<ResultT>(
            val progressFraction: Double
    ) : CountingRequestResult<ResultT>()

    data class Completed<ResultT>(
            val result: ResultT
    ) : CountingRequestResult<ResultT>()
}

CountingSink.kt

package com.example.retrofit.upload

import okhttp3.RequestBody
import okio.Buffer
import okio.ForwardingSink
import okio.Sink

typealias CountingRequestListener = (bytesWritten: Long, contentLength: Long) -> Unit

class CountingSink(
        sink: Sink,
        private val requestBody: RequestBody,
        private val onProgressUpdate: CountingRequestListener
) : ForwardingSink(sink) {
    private var bytesWritten = 0L

    override fun write(source: Buffer, byteCount: Long) {
        super.write(source, byteCount)

        bytesWritten += byteCount
        onProgressUpdate(bytesWritten, requestBody.contentLength())
    }
}

howToUse.kt

val bag = CompositeDisposable()
val uploader = AttachmentUploader()

uploader.uploadAttachment(file.name, file, "application/octet-stream")
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribeBy(
        onError = { error ->
            // Display error alert
        },
        onComplete = {
            // Display completed Snackbar
        },
        onNext = { progress ->
            // Update progress bar
        }
    )
    .addTo(bag)

server_side_express.js

const express = require('express');
const server_side = express();
const multer = require('multer');
const path = require('path');
const crypto = require('crypto');

storage = multer.diskStorage({
    destination: "path",
    filename: function(req, file, cb) {
      let extension = path.extname(file.originalname)
      // TODO: if(valid(extension)) ...
      return crypto.pseudoRandomBytes(16, function(err, raw) {
        if (err) {
          return cb(err);
        }
        return cb(null, "FILE_" + file.originalname + extension);
      });
    }
});

server_side.post(
  "/file_uploader",
  multer({
    storage: storage
  }).single('files'), function(req, res) {
    //console.log(req.body);
    return res.status(200).end();
});

server_side.listen("3000");