PipedStreams.groovy

import org.spockframework.runtime.SpockTimeoutError
import spock.lang.FailsWith
import spock.lang.Specification
import spock.lang.Timeout

import java.util.concurrent.CopyOnWriteArrayList

class PipedStreamsSpec extends Specification {

    PipedOutputStream pout = new PipedOutputStream()
    PipedInputStream pin = new PipedInputStream(pout, 3)
    BufferedOutputStream bufferedOut = new BufferedOutputStream(pout, 5)

    void "bufferSize=3のpoutに3ギリギリまで書き込んで、pin経由で3読み込む"() {
        when:
        3.times {
            pout.write(it)
        }

        then:
        3.times {
            assert pin.read() == it
        }
    }

    void "bufferSize=3のpoutに4まで書き込むと4個目の書き込みでブロックするが、1秒後にpin経由で読み込むと、pout側で4個目も書き込めて、pin側で4個全て読み込める"() {
        given:
        def readList = [] as CopyOnWriteArrayList

        when:
        Thread.startDaemon {
            4.times {
                println "Writing...: $it"
                pout.write(it)
                println "Done writing: $it"
            }
        }

        and:
        println "Waiting..."
        Thread.sleep 1_000

        and:
        4.times {
            println "Reading...: $it"
            readList << pin.read()
            println "Done Read: $it"
        }

        then:
        while (readList.size() < 4) {
            Thread.sleep 100
        }
        and:

        readList == [0, 1, 2, 3]
    }

    void "bufferSize=3のpoutに3まで書き込んだ後にクローズしても、pinから3つ読み込める"() {
        when:
        3.times {
            pout.write(it)
        }
        pout.close()

        and:
        Thread.sleep 1_000

        then:
        3.times {
            assert pin.read() == it
        }
    }

    void "bufferSize=3のpoutに3まで書き込んだ後にクローズすると、pinで正しくEOFが返せるのでEOFを期待する処理が正しく動作する"() {
        when:
        3.times {
            pout.write(it)
        }
        pout.close()

        and:
        Thread.sleep 1_000

        then:
        pin.bytes == [0, 1, 2] as byte[]
    }

    @Timeout(3)
    @FailsWith(SpockTimeoutError) //=> これがうまく機能しないのが無念
    void "bufferSize=3のpoutに3まで書き込んだ後にクローズしないまま、pinでEOFを期待する処理を実行するとEOFが返らないのでずっとブロックしてしまう"() {
        when:
        3.times {
            pout.write(it)
        }
        //pout.close()

        and:
        Thread.sleep 5_000

        then:
        pin.bytes == [0, 1, 2] as byte[]
    }

    void "poutがクローズしているときに、pinから読み込むとEOFが返る"() {
        when:
        pout.close()

        then:
        pin.read() == -1
    }

    void "poutがクローズしているときに、poutに書き込むとIOException(Pipe closed)になる"() {
        when:
        pout.close()

        and:
        pout.write(1)

        then:
        IOException e = thrown()
        e.message == "Pipe closed"
    }

    void "pinがクローズしているときに、poutに書き込むとIOException(Pipe closed)になる"() {
        when:
        pin.close()

        and:
        pout.write(1)

        then:
        IOException e = thrown()
        e.message == "Pipe closed"
    }

    void "pinがクローズしているときに、pinから読み込むとIOException(Pipe closed)になる"() {
        when:
        pin.close()

        and:
        pin.read()

        then:
        IOException e = thrown()
        e.message == "Pipe closed"
    }

    void "bufferSize=3のpoutをラップしたbufferSize=5のbufferedOutに3まで書き込んで、pin経由で3読み込む"() {
        when:
        3.times {
            bufferedOut.write(it)
            bufferedOut.flush() // flushしないとパイプまで届かないので注意
        }

        then:
        3.times {
            assert pin.read() == it
        }
    }

    void "bufferSize=3のpoutをラップしたbufferSize=5のbufferedOutに5まで書き込むと4個目のフラッシュでブロックするが、1秒後にpin経由で読み込むと、bufferedOut側のフラッシュも進んで、pin側で5個全て読み込める"() {
        given:
        def readList = [] as CopyOnWriteArrayList

        when:
        Thread.startDaemon {
            5.times {
                println "Writing...: $it"
                bufferedOut.write(it)
                println "Flushing...: $it"
                bufferedOut.flush() // flushしないとパイプまで届かないのでreadでブロックしてしまうので注意。ただし、flushによってpoutのbufferSizeによりブロックされる
                println "Done Flushing: $it"
            }
        }

        and:
        println "Waiting..."
        Thread.sleep 1_000

        and:
        5.times {
            println "Reading...: $it"
            readList << pin.read()
            println "Done Read: $it"
        }

        then:
        while (readList.size() < 5) {
            Thread.sleep 100
        }
        and:

        readList == [0, 1, 2, 3, 4]
    }

    void "bufferSize=3のpoutをラップしたbufferSize=5のbufferedOutに10まで書き込むと4個目のフラッシュでブロックするが、1秒後にpin経由で読み込むと、bufferedOut側のフラッシュも進んで、pin側で10個全て読み込める"() {
        given:
        def readList = [] as CopyOnWriteArrayList

        when:
        Thread.startDaemon {
            10.times {
                println "Writing...: $it"
                bufferedOut.write(it)
                println "Flushing...: $it"
                bufferedOut.flush() // flushしないとパイプまで届かないので注意。ただし、flushによってpoutのbufferSizeによりブロックされる
                println "Done Flushing: $it"
            }
        }

        and:
        println "Waiting..."
        Thread.sleep 1_000

        and:
        10.times {
            println "Reading...: $it"
            readList << pin.read()
            println "Done Read: $it"
        }

        then:
        while (readList.size() < 10) {
            Thread.sleep 100
        }
        and:

        readList == [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
    }
}