Tokio和serde：反序列化JSON

Question

在尝试从tokio的反序列化为JSON时，我正尝试使用这个函数：

use futures::prelude::*;
use serde_json::Value;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::stream::StreamExt;
use tokio_serde_json::ReadJson;
use tokio_util::codec::{FramedRead, LengthDelimitedCodec};

pub async fn handle_stream(mut stream: TcpStream) {
    let length_delimited = FramedRead::new(stream, LengthDelimitedCodec::new());
    let mut deserialized = ReadJson::<_, Value>::new(length_delimited);
}

我的依赖版本如下：

tokio = { version = "0.2.22", features=["full"] }
tokio-util = { version = "0.3.0", features=["full"] }
tokio-serde = "0.6"
serde = "1.0.92"
serde_derive = "1.0.92"
serde_json = "1.0.39"
futures-preview = "0.3.0-alpha"

我从 carllerche/tokio-serde-json 的示例中获取了此代码:

但是，我收到了以下错误消息：

error[E0277]: the trait bound `tokio_util::codec::framed_read::FramedRead<tokio::net::tcp::stream::TcpStream, tokio_util::codec::length_delimited::LengthDelimitedCodec>: futures::stream::Stream` is not satisfied
   --> src/kad2.rs:141:58
    |
141 |         let mut deserialized = ReadJson::<_, Value>::new(length_delimited);
    |                                                          ^^^^^^^^^^^^^^^^ the trait `futures::stream::Stream` is not implemented for `tokio_util::codec::framed_read::FramedRead<tokio::net::tcp::stream::TcpStream, tokio_util::codec::length_delimited::LengthDelimitedCodec>`
    |
    = note: required by `tokio_serde_json::ReadJson::<T, U>::new`

这是否是API和推荐库已经演变并转移到另一种操作方式的情况？现在我该如何使用Tokio将JSON反序列化为Value？

- Nicolas Marshall

Answer 1

看起来这个仓库已经被另外一个仓库所取代：

之前使用的是：tokio-serde-json

现在使用的是：tokio-serde

https://github.com/carllerche/tokio-serde/blob/master/examples/server.rs

我使用了这段服务端代码：

use futures::prelude::*;
use serde_json::Value;
use tokio::net::TcpListener;
use tokio_serde::formats::*;
use tokio_util::codec::{FramedRead, LengthDelimitedCodec};

#[tokio::main]
pub async fn main() {
    // Bind a server socket
    let listener = TcpListener::bind("127.0.0.1:17653").await.unwrap();

    println!("listening on {:?}", listener.local_addr());

    loop {
        let (socket, _) = listener.accept().await.unwrap();

        // Delimit frames using a length header
        let length_delimited = FramedRead::new(socket, LengthDelimitedCodec::new());

        // Deserialize frames
        let mut deserialized = tokio_serde::SymmetricallyFramed::new(
            length_delimited,
            SymmetricalJson::<Value>::default(),
        );

        // Spawn a task that prints all received messages to STDOUT
        tokio::spawn(async move {
            while let Some(msg) = deserialized.try_next().await.unwrap() {
                println!("GOT: {:?}", msg);
            }
        });
    }
}

以下是相关依赖项：

[dependencies]
futures = "0.3"
impls = "1"
tokio = { version = "1.0", features = ["full"] }
tokio-util = { version = "0.6", features = ["codec"] }
static_assertions = "1.1.0"
serde_json = "1.0"
tokio-serde = { version = "0.8", features = ["json"] }

现在它正在成功编译和运行。

- Amoo Hesam