Rust 100 Ex 🏃【30/37】 双方向通信・リファクタリング ～返信用封筒を入れよう！～

前の記事

• 【0】 準備 ← 初回
• ...
• 【29】 チャネル・参照の内部可変性 ～Rustの虎の子、mpscと Rc<RefCell<T>>～ ← 前回
• 【30】 双方向通信・リファクタリング ～返信用封筒を入れよう！～ ← 今回

全記事一覧

• 【0】 準備
• 【1】 構文・整数・変数
• 【2】 if・パニック・演習
• 【3】 可変・ループ・オーバーフロー
• 【4】 キャスト・構造体 (たまにUFCS)
• 【5】 バリデーション・モジュールの公開範囲 ～ → カプセル化！～
• 【6】 カプセル化の続きと所有権とセッター ～そして不変参照と可変参照！～
• 【7】 スタック・ヒープと参照のサイズ ～メモリの話～
• 【8】 デストラクタ(変数の終わり)・トレイト ～終わりと始まり～
• 【9】 Orphan rule (孤児ルール)・演算子オーバーロード・derive ～Empowerment 💪 ～
• 【10】 トレイト境界・文字列・Derefトレイト ～トレイトのアレコレ～
• 【11】 Sized トレイト・From トレイト・関連型 ～おもしろトレイトと関連型～
• 【12】 Clone・Copy・Dropトレイト ～覚えるべき主要トレイトたち～
• 【13】 トレイトまとめ・列挙型・match式 ～最強のトレイトの次は、最強の列挙型～
• 【14】 フィールド付き列挙型とOption型 ～チョクワガタ～
• 【15】 Result型 ～Rust流エラーハンドリング術～
• 【16】 Errorトレイトと外部クレート ～依存はCargo.tomlに全部お任せ！～
• 【17】 thiserror・TryFrom ～トレイトもResultも自由自在！～
• 【18】 Errorのネスト・慣例的な書き方 ～Rustらしさの目醒め～
• 【19】 配列・動的配列 ～スタックが使われる配列と、ヒープに保存できる動的配列～
• 【20】 動的配列のリサイズ・イテレータ ～またまたトレイト登場！～
• 【21】 イテレータ・ライフタイム ～ライフタイム注釈ようやく登場！～
• 【22】 コンビネータ・RPIT ～ 「 Iterator トレイトを実装してるやつ」～
• 【23】 impl Trait ・スライス ～配列の欠片～
• 【24】 可変スライス・下書き構造体 ～構造体で状態表現～
• 【25】 インデックス・可変インデックス ～インデックスもトレイト！～
• 【26】 HashMap・順序・BTreeMap ～Rustの辞書型～
• 【27】 スレッド・'staticライフタイム ～並列処理に見るRustの恩恵～
• 【28】 リーク・スコープ付きスレッド ～ライフタイムに技あり！～
• 【29】 チャネル・参照の内部可変性 ～Rustの虎の子、mpscと Rc<RefCell<T>>～
• 【30】 双方向通信・リファクタリング ～返信用封筒を入れよう！～
• 【31】 上限付きチャネル・PATCH機能 ～パンクしないように制御！～
• 【32】 Send・排他的ロック(Mutex)・非対称排他的ロック(RwLock) ～真打 Arc<Mutex<T>> 登場～
• 【33】 チャネルなしで実装・Syncの話 ～考察回です～
• 【34】 async fn・非同期タスク生成 ～Rustの非同期入門～
• 【35】 非同期ランタイム・Futureトレイト ～非同期のお作法～
• 【36】 ブロッキング・非同期用の実装・キャンセル ～ラストスパート！～
• 【37】 Axumでクラサバ！ ～最終回～
• 【おまけ1】 Rustで勘違いしていたこと3選 🏄🌴 【100 Exercises To Learn Rust 🦀 完走記事 🏃】
• 【おまけ2】 【🙇 懺悔 🙇】Qiitanグッズ欲しさに1日に33記事投稿した話またはQiita CLIとcargo scriptを布教する的な何か

100 Exercise To Learn Rust 演習第30回になります、今回は新しい機能の紹介はなく、リファクタリングが中心の回です！

今回の関連ページ

• 7.7. Ack pattern
• 7.8. Client

[07_threads/07_ack] 双方向通信・Ackパターン

• 7.7. Ack pattern

問題はこちらです。

src/lib.rs

use std::sync::mpsc::{Receiver, Sender};
use crate::store::TicketStore;

pub mod data;
pub mod store;

// Refer to the tests to understand the expected schema.
pub enum Command {
    Insert { todo!() },
    Get { todo!() }
}

pub fn launch() -> Sender<Command> {
    let (sender, receiver) = std::sync::mpsc::channel();
    std::thread::spawn(move || server(receiver));
    sender
}

// TODO: handle incoming commands as expected.
pub fn server(receiver: Receiver<Command>) {
    let mut store = TicketStore::new();
    loop {
        match receiver.recv() {
            Ok(Command::Insert {}) => {
                todo!()
            }
            Ok(Command::Get {
                todo!()
            }) => {
                todo!()
            }
            Err(_) => {
                // There are no more senders, so we can safely break
                // and shut down the server.
                break
            },
        }
    }
}

todo!() があるのは lib.rs だけですが、今回はテストの方も参照する必要がありそうです。

tests/insert.rs

use response::data::{Status, Ticket, TicketDraft};
use response::store::TicketId;
use response::{launch, Command};
use ticket_fields::test_helpers::{ticket_description, ticket_title};

#[test]
fn insert_works() {
    let sender = launch();
    let (response_sender, response_receiver) = std::sync::mpsc::channel();

    let draft = TicketDraft {
        title: ticket_title(),
        description: ticket_description(),
    };
    let command = Command::Insert {
        draft: draft.clone(),
        response_sender,
    };

    sender
        .send(command)
        // If the thread is no longer running, this will panic
        // because the channel will be closed.
        .expect("Did you actually spawn a thread? The channel is closed!");

    let ticket_id: TicketId = response_receiver.recv().expect("No response received!");

    let (response_sender, response_receiver) = std::sync::mpsc::channel();
    let command = Command::Get {
        id: ticket_id,
        response_sender,
    };
    sender
        .send(command)
        .expect("Did you actually spawn a thread? The channel is closed!");

    let ticket: Ticket = response_receiver
        .recv()
        .expect("No response received!")
        .unwrap();
    assert_eq!(ticket_id, ticket.id);
    assert_eq!(ticket.status, Status::ToDo);
    assert_eq!(ticket.title, draft.title);
    assert_eq!(ticket.description, draft.description);
}

その他のソースコード

src/data.rs

use crate::store::TicketId;
use ticket_fields::{TicketDescription, TicketTitle};

#[derive(Clone, Debug, PartialEq)]
pub struct Ticket {
    pub id: TicketId,
    pub title: TicketTitle,
    pub description: TicketDescription,
    pub status: Status,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TicketDraft {
    pub title: TicketTitle,
    pub description: TicketDescription,
}

#[derive(Clone, Debug, Copy, PartialEq, Eq)]
pub enum Status {
    ToDo,
    InProgress,
    Done,
}

src/store.rs

use crate::data::{Status, Ticket, TicketDraft};
use std::collections::BTreeMap;

#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub struct TicketId(u64);

#[derive(Clone)]
pub struct TicketStore {
    tickets: BTreeMap<TicketId, Ticket>,
    counter: u64,
}

impl TicketStore {
    pub fn new() -> Self {
        Self {
            tickets: BTreeMap::new(),
            counter: 0,
        }
    }

    pub fn add_ticket(&mut self, ticket: TicketDraft) -> TicketId {
        let id = TicketId(self.counter);
        self.counter += 1;
        let ticket = Ticket {
            id,
            title: ticket.title,
            description: ticket.description,
            status: Status::ToDo,
        };
        self.tickets.insert(id, ticket);
        id
    }

    pub fn get(&self, id: TicketId) -> Option<&Ticket> {
        self.tickets.get(&id)
    }
}

解説

tests/insert.rs の command 変数に注目すると、何やら面白そうなことをしています！

Rust

let command = Command::Insert {
    draft: draft.clone(),
    response_sender,
};

let command = Command::Get {
    id: ticket_id,
    response_sender,
};

Command は Sender 、 Receiver がメッセージをやり取りするための構造体です。その Command が更に送信用に Sender<?> を内包しています。言わば返信用封筒です！

テストを読むと、コマンドによって ? 部分は異なることがわかります。 Insert の方は TicketId 、 Get の方は見つからなかった時も考慮してか Option<Ticket> になっています。

以上を踏まえ、 server 関数内の match 式を埋めていきます。

src/lib.rs

use crate::data::{Ticket, TicketDraft};
use crate::store::{TicketId, TicketStore};
use std::sync::mpsc::{Receiver, Sender};

pub mod data;
pub mod store;

// Refer to the tests to understand the expected schema.
pub enum Command {
    Insert {
        draft: TicketDraft,
        response_sender: Sender<TicketId>,
    },
    Get {
        id: TicketId,
        response_sender: Sender<Option<Ticket>>,
    },
}

pub fn launch() -> Sender<Command> {
    let (sender, receiver) = std::sync::mpsc::channel();
    std::thread::spawn(move || server(receiver));
    sender
}

// TODO: handle incoming commands as expected.
pub fn server(receiver: Receiver<Command>) {
    let mut store = TicketStore::new();
    loop {
        match receiver.recv() {
            Ok(Command::Insert {
                draft,
                response_sender,
            }) => {
                let id = store.add_ticket(draft);
                response_sender.send(id).unwrap();
            }
            Ok(Command::Get {
                id,
                response_sender,
            }) => {
                let ticket = store.get(id).cloned();
                response_sender.send(ticket).unwrap();
            }
            Err(_) => {
                // There are no more senders, so we can safely break
                // and shut down the server.
                break;
            }
        }
    }
}

TicketStore 操作の返り値がちょうど返信用封筒に入れる型と一致しているので、そのまま返してあげています。

Book 曰く、この返信用封筒パターンはAckパターン¹と呼ばれRustにおいてよく見られるパターンだそうです。

筆者はスレッド間通信で困った時に無意識にたどり着いていたのですが、一般的なパターンだったんですね...以前書いた記事で登場していて執筆時点では「これでいいのかなぁ」と思っていたのですが、今後は積極的に推していこうと思います！

[07_threads/08_client] クライアント

• 7.8. Client

問題はこちらです。本問題もチケットサービスのクライアント・サーバーシステムのリファクタがテーマで、目新しいトピックはありません

src/lib.rs

use crate::data::{Ticket, TicketDraft};
use crate::store::{TicketId, TicketStore};
use std::sync::mpsc::{Receiver, Sender};

pub mod data;
pub mod store;

#[derive(Clone)]
// TODO: flesh out the client implementation.
pub struct TicketStoreClient {}

impl TicketStoreClient {
    // Feel free to panic on all errors, for simplicity.
    pub fn insert(&self, draft: TicketDraft) -> TicketId {
        todo!()
    }

    pub fn get(&self, id: TicketId) -> Option<Ticket> {
        todo!()
    }
}

pub fn launch() -> TicketStoreClient {
    let (sender, receiver) = std::sync::mpsc::channel();
    std::thread::spawn(move || server(receiver));
    todo!()
}

// No longer public! This becomes an internal detail of the library now.
enum Command {
    Insert {
        draft: TicketDraft,
        response_channel: Sender<TicketId>,
    },
    Get {
        id: TicketId,
        response_channel: Sender<Option<Ticket>>,
    },
}

fn server(receiver: Receiver<Command>) {
    let mut store = TicketStore::new();
    loop {
        match receiver.recv() {
            Ok(Command::Insert {
                draft,
                response_channel,
            }) => {
                let id = store.add_ticket(draft);
                let _ = response_channel.send(id);
            }
            Ok(Command::Get {
                id,
                response_channel,
            }) => {
                let ticket = store.get(id);
                let _ = response_channel.send(ticket.cloned());
            }
            Err(_) => {
                // There are no more senders, so we can safely break
                // and shut down the server.
                break;
            }
        }
    }
}

その他のファイル

tests/insert.rs

use client::data::{Status, TicketDraft};
use client::launch;
use ticket_fields::test_helpers::{ticket_description, ticket_title};

#[test]
fn insert_works() {
    // Notice how much simpler the test is now that we have a client to handle the details!
    let client = launch();
    let draft = TicketDraft {
        title: ticket_title(),
        description: ticket_description(),
    };
    let ticket_id = client.insert(draft.clone());

    let client2 = client.clone();
    let ticket = client2.get(ticket_id).unwrap();
    assert_eq!(ticket_id, ticket.id);
    assert_eq!(ticket.status, Status::ToDo);
    assert_eq!(ticket.title, draft.title);
    assert_eq!(ticket.description, draft.description);
}

src/data.rs

use crate::store::TicketId;
use ticket_fields::{TicketDescription, TicketTitle};

#[derive(Clone, Debug, PartialEq)]
pub struct Ticket {
    pub id: TicketId,
    pub title: TicketTitle,
    pub description: TicketDescription,
    pub status: Status,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TicketDraft {
    pub title: TicketTitle,
    pub description: TicketDescription,
}

#[derive(Clone, Debug, Copy, PartialEq, Eq)]
pub enum Status {
    ToDo,
    InProgress,
    Done,
}

src/store.rs

use crate::data::{Status, Ticket, TicketDraft};
use std::collections::BTreeMap;

#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub struct TicketId(u64);

#[derive(Clone)]
pub struct TicketStore {
    tickets: BTreeMap<TicketId, Ticket>,
    counter: u64,
}

impl TicketStore {
    pub fn new() -> Self {
        Self {
            tickets: BTreeMap::new(),
            counter: 0,
        }
    }

    pub fn add_ticket(&mut self, ticket: TicketDraft) -> TicketId {
        let id = TicketId(self.counter);
        self.counter += 1;
        let ticket = Ticket {
            id,
            title: ticket.title,
            description: ticket.description,
            status: Status::ToDo,
        };
        self.tickets.insert(id, ticket);
        id
    }

    pub fn get(&self, id: TicketId) -> Option<&Ticket> {
        self.tickets.get(&id)
    }
}

解説

「今までクライアント用の処理は毎回ベタ書きしていたけど、それだと非効率だよね！」ということで、クライアント専用構造体をサーバー作成段階で同時に作成してしまい、その構造体にて挿入処理や取得処理を書こう！というリファクタの問題ですね！

src/lib.rs

use crate::data::{Ticket, TicketDraft};
use crate::store::{TicketId, TicketStore};
use std::sync::mpsc::{Receiver, Sender};

pub mod data;
pub mod store;

#[derive(Clone)]
// TODO: flesh out the client implementation.
pub struct TicketStoreClient {
    command_sender: Sender<Command>,
}

impl TicketStoreClient {
    // Feel free to panic on all errors, for simplicity.
    pub fn insert(&self, draft: TicketDraft) -> TicketId {
        let (response_channel, response_receiver) = std::sync::mpsc::channel();
        self.command_sender
            .send(Command::Insert {
                draft,
                response_channel,
            })
            .unwrap();
        response_receiver.recv().unwrap()
    }

    pub fn get(&self, id: TicketId) -> Option<Ticket> {
        let (response_channel, response_receiver) = std::sync::mpsc::channel();
        self.command_sender
            .send(Command::Get {
                id,
                response_channel,
            })
            .unwrap();
        response_receiver.recv().unwrap()
    }
}

pub fn launch() -> TicketStoreClient {
    let (command_sender, command_receiver) = std::sync::mpsc::channel();
    std::thread::spawn(move || server(command_receiver));
    TicketStoreClient { command_sender }
}

// 以降は改変なし

前のエクササイズのテスト等を参考に、3箇所ある todo!() を埋めていきます。メソッドの定義と launch で返す値の記述だけですね！

その他一つだけ抑えておきたい点としては、 TicketStoreClient に #[derive(Clone)] が付いていることでしょうか。サーバー作成時にクライアントを一つしか返しておらず、一見すると一つしかクライアントを持てないように見えますが、もしクライアントを複数スレッドに分けたいとなったら、複製できるのでこれで大丈夫です！

では次の問題に行きましょう！

次の記事: 【31】 上限付きチャネル・PATCH機能 ～パンクしないように制御！～

1. TCPのSYN/ACKと同様、"Acknowledgement" (了承)の略語と思われます。 ↩