refCount维持BLE连接出错后仍保持连接、重放错误的解决方法咨询

问题根源 #

当前使用的replay+refCount组合存在固有特性：上游BLE连接流发送onError终止事件后，replay内部缓存会永久持有该错误事件，所有后续新订阅都会直接收到重放的错误；同时refCount接收到终止事件后不会自动重置共享状态，因此会出现连接被持续持有的假象。

方案一：出错后立即断开连接（对应第一个需求） #

核心逻辑是错误发生时主动终止当前共享流，直接释放上游连接，同时重置refCount的共享状态，下次新订阅时会自动重建连接。

修改后代码：

bleDevice
    .establishConnection(false)
    // 捕获上游错误，转成onComplete事件终止当前流，触发refCount释放连接
    .onErrorResumeNext(Observable.empty())
    .replay(1)
    .refCount(1, 5, TimeUnit.MINUTES)
    // 可选：如果需要订阅侧感知到连接错误，可以加上下行注释的代码，自定义异常抛出
    // .switchIfEmpty(Observable.error(new BLEConnectionBrokenException()))

该方案适合不需要自动重连的场景，错误发生后5分钟保活逻辑也会同步终止，不会再持有无效连接。

方案二：禁止replay重放错误（对应第二个需求） #

核心逻辑是将错误事件转换为普通事件处理，避免replay将错误识别为终止事件进行缓存。

如果需要当前订阅者能收到错误、新订阅者不重放历史错误，代码如下：

bleDevice
    .establishConnection(false)
    // 将所有事件包装为普通Notification对象，避免replay识别到终止事件
    .materialize()
    .replay(1)
    .refCount(1, 5, TimeUnit.MINUTES)
    // 下游将Notification还原为正常事件传递给订阅者
    .dematerialize(Function.identity())

如果需要完全不缓存错误，只重放正常的连接数据，调整代码增加过滤逻辑即可：

bleDevice
    .establishConnection(false)
    .materialize()
    // 仅允许正常的业务数据进入缓存，错误/完成事件直接丢弃不缓存
    .filter(Notification::isOnNext)
    .replay(1)
    .refCount(1, 5, TimeUnit.MINUTES)
    .dematerialize(Function.identity())

该方案适合需要保留5分钟保活逻辑，仅需要屏蔽错误重放的场景。

内容的提问来源于stack exchange，提问作者RefuX