/*
    getByteStream()
        .observeOn(AndroidSchedulers.mainThread())
        .subscribeOn(Schedulers.computation())
        .writeBytesToFile(getPath(), generateNewFileName())
        .subscribeBy(onError = renderError())
        .addTo(disposables)
 */

fun Flowable<ByteArray>.writeBytesToFile(filesDir: File, fileName: String): Flowable<ByteArray> {
    val resourceSupplier = FileOutputStreamSupplier(filesDir, fileName)

    val sourceSupplier = Function<FileOutputStream, Flowable<ByteArray>> { fileOs ->
        this@writeBytesToFile.doOnNext { byteArray ->
            fileOs.write(byteArray)
        }
    }

    val disposer = Consumer<FileOutputStream> {
        it.flush()
        it.close()
    }

    return Flowable.using(resourceSupplier, sourceSupplier, disposer)
}

internal class FileOutputStreamSupplier(private val parent: File,
                               private val fileName: String) : Callable<FileOutputStream> {
    override fun call(): FileOutputStream {
        parent.mkdirs()
        val file = File(parent, fileName)
        return FileOutputStream(file)
    }
}