stream.kt

import io.grpc.ManagedChannelBuilder
import io.grpc.stub.StreamObserver
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.launch
import org.apache.commons.compress.archivers.tar.TarArchiveEntry
import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream
import java.io.*
import java.nio.file.Files
import java.nio.file.Path
import java.nio.file.Paths

// Import your generated gRPC classes here
// import your.package.YourGrpcServiceGrpc
// import your.package.UploadRequest
// import your.package.UploadResponse

fun tarDirectoryToStream(directory: Path, outputStream: OutputStream) {
    TarArchiveOutputStream(outputStream).use { out ->
        Files.walk(directory).forEach { path ->
            if (Files.isRegularFile(path)) {
                val entry = TarArchiveEntry(path.toFile(), directory.relativize(path).toString())
                out.putArchiveEntry(entry)
                Files.copy(path, out)
                out.closeArchiveEntry()
            }
        }
        out.finish()
    }
}

fun uploadTarballToGrpc(inputStream: InputStream) {
    val channel = ManagedChannelBuilder.forAddress("localhost", 50051).usePlaintext().build()

    // Assuming your gRPC service is called YourGrpcService and has a method called uploadFile
    val stub = YourGrpcServiceGrpc.newStub(channel)

    val requestObserver = stub.uploadFile(object : StreamObserver<UploadResponse> {
        override fun onNext(value: UploadResponse) {
            println("Upload response: ${value.status}")
        }

        override fun onError(t: Throwable) {
            t.printStackTrace()
        }

        override fun onCompleted() {
            println("Upload completed")
            channel.shutdown()
        }
    })

    try {
        val buffer = ByteArray(4096)
        var bytesRead: Int
        while (inputStream.read(buffer).also { bytesRead = it } != -1) {
            val request = UploadRequest.newBuilder()
                .setFileChunk(com.google.protobuf.ByteString.copyFrom(buffer, 0, bytesRead))
                .build()
            requestObserver.onNext(request)
        }
    } catch (e: Exception) {
        requestObserver.onError(e)
    } finally {
        requestObserver.onCompleted()
    }
}

fun main() {
    val directoryToTar = Paths.get("path/to/your/directory")

    val pipedInputStream = PipedInputStream()
    val pipedOutputStream = PipedOutputStream(pipedInputStream)

    GlobalScope.launch(Dispatchers.IO) {
        try {
            tarDirectoryToStream(directoryToTar, pipedOutputStream)
        } catch (e: Exception) {
            e.printStackTrace()
        } finally {
            pipedOutputStream.close()
        }
    }

    uploadTarballToGrpc(pipedInputStream)
}

dependencies {
    implementation("org.apache.commons:commons-compress:1.21")
    implementation("io.grpc:grpc-netty-shaded:1.42.1")
    implementation("io.grpc:grpc-protobuf:1.42.1")
    implementation("io.grpc:grpc-stub:1.42.1")
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.6.0")
}