Rustで巨大な配列データやネットワークストリームをファイルに書き込みながらシリアライズする方法

Question

やりたいこと

ネットワークからメッセージを受信した瞬間に受信したデータをメモリ内にバッファリングせずJSONでシリアライズし、即時吐き出したい場合はどうしたら良いでしょう。

受信したメッセージは配列の一要素として吐き出したいです。
（ログを一つのJSONオブジェクトとして吐き出すようなイメージです。）

雰囲気（ビルドエラーになります）

struct SeqSerializer{
    seq: Compound<'a, &'a mut dyn Write, CompactFormatter>,
}
impl SeqSerializer {
    // 初期化
    fn new() -> Self {
        let writer = std::io::stdout();
        let mut seq = serde_json::Serializer::new(writer);
        let mut seq = seq.serialize_seq(None).unwrap();
        SeqSerializer { seq: seq } // ↑がキャプチャしてるので返せない。
    }
    // 通信がオープンした
    fn open(&self) {}
    // メッセージを受信した
    fn receive_message(&self, msg: String) {
        seq.serialize_element(msg);
    }    // 通信がクローズされた
    fn close(&self) {
        seq.end();
    }
}

依存関係は以下のとおりです。

[dependencies]
serde_derive = "1.0.115"
serde = "1.0.115"
serde_json = "1.0.57"

コメントで記載した通り、seq._serialize_seqがキャプチャしてしまうため、意図したコードを書くことができません。
こういった
「初期化」
「使用」
「後処理」
をバラバラに使う場合、どのようにすれば書くことができるのでしょうか。

それともこのような使い方は、そもそもおかしいからやめろということでしょうか。

Answer 1

前の回答のように SeqSerializerを使用する関数を渡す形にできず、どうしても初期化でSeqSerializerを返す形にしたい場合は、処理を全て非同期ランタイム上で行うという手もあります。

ただし、この方法は処理が複雑になってくると対応しきれなくなるかもしれません。（例えば、前の回答は同様の手法を二重に組み合わせることも容易ですが、こちらはかなり難しくなります。）

[dependencies]
serde_derive = "1.0.115"
serde = "1.0.115"
serde_json = "1.0.57"
tokio = { version = "1.18.2", features = [
    "sync",
    "rt",
    "macros",
    "rt-multi-thread",
] }

use serde::{ser::SerializeSeq, Serializer};
use serde_json::Result;
use tokio::{
    spawn,
    sync::{mpsc, oneshot},
    task::JoinHandle,
};

struct SeqSerializer {
    sender: mpsc::UnboundedSender<(String, oneshot::Sender<Result<()>>)>,
    handle: JoinHandle<Result<()>>,
}
impl SeqSerializer {
    fn open() -> Self {
        let (sender, mut reciever) =
            mpsc::unbounded_channel::<(String, oneshot::Sender<Result<()>>)>();
        let handle = spawn(async move {
            let writer = std::io::stdout();
            let mut seq = serde_json::Serializer::new(writer);
            let mut seq = seq.serialize_seq(None).unwrap();
            while let Some((msg, sender)) = reciever.recv().await {
                sender.send(seq.serialize_element(&msg)).unwrap();
            }
            seq.end()
        });
        Self { sender, handle }
    }
    async fn receive_message(&self, msg: String) -> Result<()> {
        let (sender, receiver) = oneshot::channel();
        self.sender.send((msg, sender)).unwrap();
        receiver.await.unwrap()
    }
    async fn close(self) -> Result<()> {
        drop(self.sender);
        self.handle.await.unwrap()
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    let s = SeqSerializer::open();
    s.receive_message("abc".into()).await?;
    s.receive_message("def".into()).await?;
    s.close().await?;
    Ok(())
}

Answer 2

初期化の処理を

「SeqSerializerを返す」

から

「SeqSerializerを使用する関数を渡す」

形に変更すると比較的少ない変更で対応できます。

use serde::{ser::SerializeSeq, Serializer};
use serde_json::ser::{CompactFormatter, Compound};
use std::io::Write;

struct SeqSerializer<'a> {
    seq: Compound<'a, &'a mut dyn Write, CompactFormatter>,
}
impl<'a> SeqSerializer<'a> {
    // 初期化
    fn new_with(f: impl FnOnce(SeqSerializer)) {
        let mut writer = std::io::stdout();
        let writer: &mut dyn Write = &mut writer;
        let mut seq = serde_json::Serializer::new(writer);
        let seq = seq.serialize_seq(None).unwrap();
        f(SeqSerializer { seq })
    }
    // 通信がオープンした
    fn open(&self) {}
    // メッセージを受信した
    fn receive_message(&mut self, msg: &str) {
        self.seq.serialize_element(msg).unwrap();
    } // 通信がクローズされた
    fn close(self) {
        self.seq.end().unwrap();
    }
}

fn main() {
    SeqSerializer::new_with(|mut s| {
        s.open();
        s.receive_message("abc");
        s.receive_message("def");
        s.close();
    });
}

Answer 3

愚直ですが、出したいデータのトップレベルが配列と確定しているなら、1エントリごとにJSONにエンコードして吐き出すという方法があります。

use std::fs::File;
use std::io::{BufWriter, Write};

use anyhow::Result;
use serde::Serialize;
use serde_json::to_writer;

fn main() -> Result<()> {
    let f = File::create("sample.json")?;
    let mut f = BufWriter::new(f);

    writeln!(f, "[")?;

    for i in 0..100 {
        if i > 0 {
            writeln!(f, ",")?; // 前のエントリの後端のカンマ
        }

        let value = receive_data();
        to_writer(&mut f, &value)?;
    }

    write!(f, "\n]")?;

    Ok(())
}

#[derive(Serialize)]
pub struct SampleStruct {
    name: String,
    age: u64,
}

fn receive_data() -> SampleStruct {
    SampleStruct {
        name: "Douglas Adams".to_string(),
        age: 42,
    }
}