PipedInputStream / PipedOutputStream を Kotlin で使う

ときに、 巨大になる可能性のある文字列とか画像（バイナリデータ）を返したいなどの理由により、 関数が返す値として InputStream を使いたくなることがある。

そんなときは PipedInputStream と PipedOutputStream を使えば解決できるのだが、 そのとき PipedInputStream を読む処理と PipedOutputStream を書く処理は 別々のスレッドでなければいけない、という制約がある。

Javaなら普通に Thread をつくって対処してもよいのであろうが、 Kotlinには Coroutines があるので、それを使って解決する方法を調べた。

環境

$ kotlin -version
Kotlin version 1.8.10-release-430 (JRE 17.0.9+9-Ubuntu-122.04)

一番簡単な例

hello.main.kts

@file:Repository("https://repo1.maven.org/maven2/")
@file:DependsOn("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.7.3")

import kotlinx.coroutines.*
import java.io.*

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

// 別スレッドにする必要がある.
GlobalScope.launch {
    PrintWriter(OutputStreamWriter(pipedOutputStream, Charsets.UTF_8), true).use {
        it.println("Hello,World!")
    }
}

val text = pipedInputStream.bufferedReader(Charsets.UTF_8).use {
    it.readText()
}

print( text )

実行します。

$ kotlin hello.main.kts
Hello,World!

できました。

より実際的な例

それでは、本題である「大きなテキストデータを返す可能性がある関数の戻値を InputStream とする例」をつくります。

まずただ単純に String を返すだけの toText() 関数を考えます。

val toText: ()->String = {
    val nameList = listOf("foo", "bar", "hoge")
    val text = nameList.joinToString("\n")
    text
}

println(toText())

実行すると、

foo
bar
hoge

が標準出力されます。

この toText() 関数が巨大な文字列を返す可能性がある場合に備えて、 この関数の戻値を InputStream にします。

val toInputStream: ()->InputStream = {
   ...
}

toText() の代わりに toInputStream() を使うコードは次のようになります。

val text = toInputStream().bufferedReader(Charsets.UTF_8).use {
    it.readText()
}

print(text)

テキスト返すのが前提ならば はじめから InputStreamReader を返すべきでは？ ともいえますが、例なのでよいことにします。

肝心の toInputStream() 関数の実装は、冒頭の Piped と Coroutines を使って 次のようになりました。

val toInputStream: ()->InputStream = {
    val pipedInputStream = PipedInputStream()
    val pipedOutputStream = PipedOutputStream(pipedInputStream)
    
    GlobalScope.launch {
        PrintWriter(OutputStreamWriter(pipedOutputStream, Charsets.UTF_8), true).use {
            val nameList = listOf("foo", "bar", "hoge")
            it.print( nameList.joinToString("\n") )
        }
    }

    pipedInputStream
}

まとめ

完成したコードを掲載します。

@file:Repository("https://repo1.maven.org/maven2/")
@file:DependsOn("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.7.3")

import kotlinx.coroutines.*
import java.io.*

/*
val toText: ()->String = {
    val nameList = listOf("foo", "bar", "hoge")
    val text = nameList.joinToString("\n")
    text
}

println(toText())
*/

val toInputStream: ()->InputStream = {
    val pipedInputStream = PipedInputStream()
    val pipedOutputStream = PipedOutputStream(pipedInputStream)
    
    GlobalScope.launch {
        PrintWriter(OutputStreamWriter(pipedOutputStream, Charsets.UTF_8), true).use {
            val nameList = listOf("foo", "bar", "hoge")
            it.print( nameList.joinToString("\n") )
        }
    }

    pipedInputStream
}

val text = toInputStream().bufferedReader(Charsets.UTF_8).use {
    it.readText()
}

print(text)

これがあれば、 大きさがわからない（巨大になるかもしれない）テキストデータや バイナリデータを返す関数をアウトオブメモリエラーなどが起きないかを心配しないで実装できそうです。

追伸

肝心の処理を別の関数として実装したい場合は suspend を使う。

この部分、

    GlobalScope.launch {
        PrintWriter(OutputStreamWriter(pipedOutputStream, Charsets.UTF_8), true).use {
            val nameList = listOf("foo", "bar", "hoge")
            it.print( nameList.joinToString("\n") )
        }
    }

今は単なる例なので、GlobalScope launch ブロックに直接書いていますが、 この部分がそれなりのコード分量になることがあります。

その場合は、suspend をつけた関数を定義する。

suspend fun doIt(pipedOutputStream:PipedOutputStream): Unit {
    PrintWriter(OutputStreamWriter(pipedOutputStream, Charsets.UTF_8), true).use {
        val nameList = listOf("foo", "bar", "hoge")
        it.print( nameList.joinToString("\n") )
    }
}

この doIt を使う側はつぎのように修正。

val toInputStream: ()->InputStream = {
    val pipedInputStream = PipedInputStream()
    val pipedOutputStream = PipedOutputStream(pipedInputStream)
    
    GlobalScope.launch {
        doIt(pipedOutputStream)
    }

    pipedInputStream
}

以上です。