Kotlin Coroutines SharedFlow collector lost some data with gRPC Kotlin

Question

I tried to use SharedFlow with gRPC Server side stream.

var sharedFlow: SharedFlow<T>? = null

fun createStream(): SharedFlow<T> {
  val stub = ServiceGrpcKt.Stub(channel)
  return stub.stream(request, header).shareIn(coroutineScope, SharingStarted.Eagerly).also {
    sharedFlow = it
  }
}

fun observeStream(): SharedFlow<T> {
  var stream = sharedFlow
  if (stream == null) {
    stream = createStream()
  }

  return stream
}

I get server side stream data by call observeStream().

The shared flow obtained by calling observeStream does not collect data intermittently.

When data is lost, it is when the collector is attached outside createStream().

I don't know why this happens and I don't know how to solve it. Please help me. Thanks.

I think it's because the collector takes a long time, so I simply received the data and changed it to do println, but it's lost.

I tested it as below because I was wondering if collect itself was not possible.

var sharedFlow: SharedFlow<T>? = null

fun createStream(): SharedFlow<T> {
  val stub = ServiceGrpcKt.Stub(channel)
  return stub.stream(request, header).shareIn(coroutineScope, SharingStarted.Eagerly).also {
    sharedFlow = it

    coroutineScope.launch {
      it.collect { data ->
        println(data)
      }
    }
  }
}

fun observeStream(): SharedFlow<T> {
  var stream = sharedFlow
  if (stream == null) {
    stream = createStream()
  }

  return stream
}

I attached the collector directly from the side that generates stream as above, but the data is not lost here.

Also, I checked which flow the collector refers to when running. I was referring to the same flow.

Answer 1

I think what you want is MutableSharedFlow - something like:

val sharedFlow = MutableSharedFlow<T>()

val stub = ServiceGrpcKt.Stub(channel)
CoroutineScope(Executors.newSingleThreadExecutor().asCoroutineDispatcher()).launch {
    sharedFlow.emitAll(stub.stream(request, header))
}

// then read from the `sharedFlow` in the usual Flow ways.