Skip operation

[asubb]

Sometimes it might be convenient to shift the beginning of the stream by skipping certain amount of time.

Reference implementation for Sample type for finite stream, with added proper sample measurement it would be easy to generalize the implementation:

@Serializable
class SkipStreamParams(
    val timeToSkip: Long,
    val timeUnit: TimeUnit
) : BeanParams()

class SkipStream(
    override val input: FiniteStream<Sample>,
    override val parameters: SkipStreamParams,
) : AbstractOperationBeanStream<Sample, Sample>(input), SinglePartitionBean, SingleBean<Sample>, FiniteStream<Sample> {

    override fun operationSequence(input: Sequence<Sample>, sampleRate: Float): Sequence<Sample> {
        val toSkipSamplesCount = timeToSampleIndexCeil(parameters.timeToSkip, parameters.timeUnit, sampleRate).toInt()
        return input.drop(toSkipSamplesCount)
    }

    override fun samplesCount(): Long = input.samplesCount()

    override fun length(timeUnit: TimeUnit): Long = input.length()
}

fun FiniteStream<Sample>.skip(timeToSkip: Long, timeUnit: TimeUnit): FiniteStream<Sample> =
    SkipStream(this, SkipStreamParams(timeToSkip, timeUnit))