将对象写入flux

Question

我想在flux中添加一个对象，这样订阅者就会被触发。这就是我到目前为止所得到的，但是write不是一个函数：

@Controller
class LobbyRSocketController() {

        data class Lobby(val player_1_token: Int, val player_2_token: Int, val game_id: Int) {
@Autowired lateinit var game: Game }

    data class Header(val playerToken: Int)
    val playersInQueue = HashMap<Int, Flux<Int>>()
    val lobbies = ArrayList<Lobby>()

    @MessageMapping("matchmaking")
    fun joinMatchmaking(header: Header): Flux<Int> {
        val playerToken = header.playerToken
        if (playersInQueue.containsKey(playerToken)) return Flux.just(-1)
        val flux = Flux.empty<Int>()
        playersInQueue[playerToken] = flux
        matchmake()
        return flux
    }

    fun matchmake() {
        if (lobbies.size >= 2) {
            val iter = playersInQueue.iterator()
            val gameId = IdGenerator.nextGameId
            val player1 = iter.next()
            val player2 = iter.next()
            lobbies.add(Lobby(player1.key, player2.key, gameId))
            player1.value.write(gameId)
            player2.value.write(gameId)
        }
    }
}

Answer 1

这是我将使用的：

fun main(args: Array<String>) {
    val processor = DirectProcessor.create<Int>().serialize()
    val sink = processor.sink()
    val flux = processor.map { x -> x }

    flux.subscribe({ x: Int -> System.out.println(x)})

    sink.next(1)
    sink.next(2)
    sink.next(5)
}