从 micronaut ServerWebSocket 发送异步消息

Question

我正在尝试使用 Micronaut（和 Groovy）实现简单的 WebSocket 服务器。我是 Micronaut 的新手，所以边走边学，而且似乎需要一些 RxJava。所以也学习了。

我可以使基本示例工作：

@CompileStatic
@ServerWebSocket("/ws")
class TimeseriesWebSocket {
    @OnOpen
    public Publisher<String> onOpen(WebSocketSession session) {
        println("opening connection")
        return session.send("Hello")
    }

    @OnMessage
    public Publisher<String> onMessage(String message, WebSocketSession session) {
        println("message received")
        return session.send("Thanks for the message")
    }

    @OnClose
    public Publisher<String> onClose(WebSocketSession session) {
        println("Closing")
        //this seems not to be delivered, I guess due to session being closed already
        return session.send("OK")
    }
}

所以这会打印出我放在那里的所有消息，并且可以在客户端连接时正常工作。客户端还会看到由 session.send(..) 返回的“Hello”和“Thanks for the message”消息。

现在我的问题是，我试图在这些方法之外发送消息。像这样：

@CompileStatic
@ServerWebSocket("/ws")
class MyWebSocket {
    @OnOpen
    public Publisher<String> onOpen(WebSocketSession session) {
        println("opening connection")
        startPing()
        return session.send("Hello")
    }

 //...(same as above)

    public void startPing(WebSocketSession session) {
        PingPing ping = new PingPing(session)
        ScheduledExecutorService exec = Executors.newSingleThreadScheduledExecutor();
        exec.scheduleAtFixedRate(ping, 0, 1, TimeUnit.SECONDS);
    }
}

class PingPing {
    WebSocketSession session

    public PingPing(WebSocketSession session) {
        this.session = session
    }

    @Override
    void run() {
        println("pinging..")
        try {
            session.send("Ping...")
        } catch (Exception e) {
            e.printStackTrace()
        }
    }
}

这会执行，但客户端上没有任何显示。现在，如果我将 session.send() 更改为 session.sendSync() 它工作正常。 ping 已发送。

send() 签名实际上是

default <T> Publisher<T> send(T message)

起初我想我应该将 Publisher 提供给其他来源以便发送。但我想情况并非如此。我意识到它有点像 Future 对象，所以如果我自己订阅它：

def publisher = session.send("Ping...")
publisher.subscribe(new Subscriber<GString>() {
    @Override
    void onSubscribe(Subscription s) {
        println("subscription")
    }

    @Override
    void onNext(GString gString) {
        println("next")
    }

    @Override
    void onError(Throwable t) {
        println("error")
        t.printStackTrace()
    }

    @Override
    void onComplete() {
        println("complete")
    }
})
println("publisher: ${publisher}")

运行上面这段代码（带有订阅），我猜它会触发当前线程上的 session.send() 并返回一个结果。但我实际上应该在哪里称呼它？在什么线程上？我查看了 RxJava 调度程序，但不太明白从哪里调用它。

再者，上面运行的结果，实际上是把消息传递给了客户端，但也抛出了一个错误：

error
io.reactivex.exceptions.MissingBackpressureException: create: could not emit value due to lack of requests
    at io.reactivex.internal.operators.flowable.FlowableCreate$ErrorAsyncEmitter.onOverflow(FlowableCreate.java:438)
    at io.reactivex.internal.operators.flowable.FlowableCreate$NoOverflowBaseAsyncEmitter.onNext(FlowableCreate.java:406)
    at io.micronaut.http.netty.websocket.NettyRxWebSocketSession.lambda$null$2(NettyRxWebSocketSession.java:191)
    at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:577)

这指的背压和值/请求是什么，我应该处理异步发送消息的实际方式是什么？我希望它只发送我要发送的单个项目..

Micronaut API 文档提到密切关注 javax.websocket API，但 javax.websocket async API 似乎更有意义，只是提供一个未来来听。

所以简短的问题是，如何使用 Micronaut Websocket API 在 Micronaut 提供的功能之外以异步模式发送消息？还是我做错了？

似乎我可能对此 API 做出了一些普遍错误的假设，但无法从文档中弄清楚，也找不到示例。

Answer 1

我在使用 micronaut v1.3.2 时遇到了同样的问题。

我设法让它使用 sendAsync/sendSync 而不是 send。

我查看了 NettyRxWebSocketSession 的实现，看起来 send 的实现方式与 sendAsync 有所不同。不清楚是因为配置还是发送的实现问题。

我希望这会有所帮助。