tokio-util 0.7.20 的 Framed 调用 sink_map_err 报 E0283
LengthDelimitedCodec 新增切片编码实现后,同一 Framed 有两个 Sink 输入类型。用 SinkExt<Bytes> 的限定调用消除歧义,保留帧协议设置。
升级 tokio-util 0.7.20 后,原先能编译的代码在 sink_map_err、flush 或 close 附近报 E0283。给变量补成 Framed<Io, LengthDelimitedCodec> 仍然无效,因为未知的不是这个 Framed 类型,而是它通过哪一种 Sink<Item> 调用扩展方法。
0.7.20 为 LengthDelimitedCodec 新增 Encoder<&[u8]>,原有 Encoder<Bytes> 仍保留。同一个 Framed 因而可能接收 Bytes,也可能接收字节切片;不带消息参数的组合器无法从调用中选出一个 Item。
为什么 send 能推断,前面的组合器却不行
send 与 feed 接收实际值,可以从 Bytes 或切片确定输入类型。sink_map_err 只是改变错误,flush、close 不带消息,这些位置没有同样的推断线索。
典型问题形态是先构造,再立即变换错误:
let sink = Framed::new(io, LengthDelimitedCodec::new())
.sink_map_err(MyError::from);
这里的 MyError 是应用已有的错误类型。给 sink_map_err 填方法级泛型,通常只能指定映射错误与闭包类型,不能替代扩展 trait 上的 Item。也不需要为这个错误去调 TCP 超时或改变长度前缀。
在调用处明确选用 Bytes
如果原协议代码本来发送 Bytes,可用完全限定调用保留这一路径。下面是独立函数片段,工程需要启用 tokio-util 的 codec 功能以及 futures-util 的 sink 支持:
use bytes::Bytes;
use futures_util::{Sink, SinkExt};
use std::io;
use tokio::io::{AsyncRead, AsyncWrite};
use tokio_util::codec::{Framed, LengthDelimitedCodec};
fn make_sink<T>(io: T) -> impl Sink<Bytes, Error = String>
where
T: AsyncRead + AsyncWrite,
{
let framed = Framed::new(io, LengthDelimitedCodec::new());
<Framed<T, LengthDelimitedCodec> as SinkExt<Bytes>>::sink_map_err(
framed,
|error: io::Error| error.to_string(),
)
}
示例为便于展示使用 String 作为映射后错误;实际项目应保留原来的错误类型与因果信息,不必把所有 I/O 错误都改成字符串。若 codec 原本配置了最大帧长、字节序或长度字段偏移,也要把原配置带入,不能因为复制 new() 恢复编译就重置线协议。
关键是 as SinkExt<Bytes>,它选择 trait 的 Item,而不只是给 Framed 补全类型。实际发送切片的项目可以选择切片路径,但需要保持借用数据在异步操作中的生命周期,不能为绕过错误任意改成悬空借用或无限期保留缓存。
若只是某次 flush 无法推断,也可以在那个调用上明确同一种 SinkExt Item。整条适配链应一致,避免前面固定成切片、后面又期望发送 Bytes。
编译通过后验证帧边界
此变化来自 0.7.20 同时存在的编码实现;0.7.19 只有 Bytes 的实现,类型推断更容易唯一确定。退回旧版本虽然可能恢复编译,也会撤掉其他变更,不能把它当作无需审查的长期处理。
更新调用后,以原来的空消息、普通消息和超过最大长度的消息覆盖编码与错误分支,核对对端收到的长度和内容。检查原 sink_map_err 适配仍保留需要的错误信息,以及正常 flush、close 的结果没有被吞掉。
这里修的是输入类型歧义,不是网络死锁。若 E0283 已消失却仍有传输失败,应继续按真实错误处理;保留调用限定与依赖锁的差异,回退时同步恢复,不修改 codec 的帧格式来掩盖类型问题。


