csv_stream.kt

import com.fasterxml.jackson.databind.MappingIterator
import com.fasterxml.jackson.dataformat.csv.CsvMapper
import com.fasterxml.jackson.dataformat.csv.CsvParser
//import com.fasterxml.jackson.dataformat.csv.CsvSchema
import java.io.File
import java.util.*
import java.util.function.Consumer
import java.util.stream.Stream
import java.util.stream.StreamSupport

class MSpliterator<T>(val src: MappingIterator<T>): Spliterator<T> {
    override fun characteristics(): Int =
        Spliterator.ORDERED or Spliterator.NONNULL or Spliterator.IMMUTABLE
    override fun estimateSize(): Long = Long.MAX_VALUE
    override fun trySplit(): Spliterator<T>? = null

    override fun tryAdvance(action: Consumer<in T>?): Boolean =
        when {
            src.hasNextValue() -> {
                action!!.accept(src.nextValue())
                true
            }
            else -> false
        }
}

fun <T> MappingIterator<T>.stream(): Stream<T> =
        StreamSupport.stream(MSpliterator(this), false)


data class Item(
        val first: String,
        val second: String
)

fun main(args: Array<String>) {

    val mapper = CsvMapper().apply {
        enable(CsvParser.Feature.WRAP_AS_ARRAY)
    }
//    val schema = CsvSchema.emptySchema().withoutHeader().withColumnSeparator(';')
    val src = File(args[1])
    val it = mapper.readerWithSchemaFor(Item::class.java).readValues<Item>(src)
    it.stream()
            .parallel()
            .map { it.first + it.second }
            .skip(3).limit(10)
            .forEach(::println)
}