I am working on creating some kafka extension functions and I want to take a `KStream<Either<E...

# arrow

[Carter Youngblood]

I am working on creating some kafka extension functions and I want to take a KStream<Either<Error, Result>> and create a Pair of KStream<Error> and KStream<Result> . I was not sure how to accomplish this using Either from arrow. We have some helpers in scala that do this now, and for my learning I wanted to do the same in kotlin - in scala it looks something like this -

def forkValuesEither: (KStream[K, LV], KStream[K, RV]) = (
      kStream.flatMapValues(_.left.toSeq),
      kStream.flatMapValues(_.right.toSeq)
    )

but I was not sure how to get to the value in arrow/kotlin of the either to return.

[raulraja]

What is KStream and does it have an extension for Traverse or Foldable?

[raulraja]

is this from the Kafka Streams libs?

[Carter Youngblood]

yes - from kafka streams api

[Carter Youngblood]

not sure on the Traverse/Foldable question..

[simon.vergauwen]

Looks like the same code should also work for Kotlin/Arrow

[simon.vergauwen]

What is the signature of flatMapValues?

[Carter Youngblood]

def flatMapValues[VR](mapper: V => Iterable[VR]): KStream[K, VR]

[Carter Youngblood]

I was not able to do the exact same - let me try again and see what the error is

[simon.vergauwen]

Oh nevermind, it’s the _.left.toSeq that threw me off

[simon.vergauwen]

It’s doing leftProjection and then toSeq which is

[simon.vergauwen]

fun <A, B> Either<A, B>.leftIterable(): List<A> = when(this) {
   is Either.Left -> listOf(l)
   is EIther.Right -> emptyList()
}

[simon.vergauwen]

Which means you can use flatMapValues to filter left and right using Seq but that’s a little hacky IMO 😄

[simon.vergauwen]

So maybe you can also do something like this

[Carter Youngblood]

I think thats exactly what the scala leftProjection is doing

[simon.vergauwen]

fun <A, B> KStream<A>.mapFilter(f: (A) -> B?): KStream<B> = ...

[Carter Youngblood]

this is the toSeq on the LeftProjection from scala

def toSeq: Seq[A] = e match {
      case Left(a) => Seq(a)
      case _       => Seq.empty
    }

[simon.vergauwen]

And then you could do

kStream.mapFilter { (it as? Left)?.l }

[simon.vergauwen]

let me write a complete snippet 😉 I’m not familiar with Kafka tho

[Carter Youngblood]

thanks! - I think your suggestions are already getting me further 😉

[Carter Youngblood]

kafka knowledge or not - I think you see/understand the types better than I

[simon.vergauwen]

Here you go

[simon.vergauwen]

I mocked KStream using List but all the signatures and functions are there

Untitled.kt

[simon.vergauwen]

Was a bit confusing originally with the left projection in Scala 😄 took me a second

[Carter Youngblood]

@simon.vergauwen - what a huge help, thanks for this! Really can’t say enough about Arrow and its community! Thanks for taking the time!

[simon.vergauwen]

Glad I was able to help 👍