Spring Cloud Stream доступ к сыром потоку

Question

Вот мой пример использования: пользователь подписывается на мой поток с помощью websocket (GraphQl с подпиской), мне нужно вернуть экземпляр org.reactivestreams.Publisher (который должен быть моей подпиской kafka topi c) по идентификатору пользователя.

Для иллюстрации примерно так:

/ **
  *  I don´t know how to get a instance of Publisher<Balance>
  *  It should be a consumer from a kafka topic
  */
fun balance(myStream: Publisher<Balance>, userId: String): Publisher<Balance> {
    return myStream.filter { it.userId == userId }
}

Answer 1

Может быть, вам нужно написать потребителя Spring Cloud Stream и затем опубликовать sh его программно в WebSocket. Что-то вроде

public Consumer<Flux<Balance>> myStream() {

  //filter here and then publish to websocket.

}

Здесь является примером реализации приемника WebSocket, которую вы можете использовать в качестве руководства, но это не реагирует.