Rust에서 HTTP Long Polling을 구현하는 방법

tokio's team : axum에서 개발한 새로운 웹 프레임워크를 사용합니다. 그 성능과 단순성은 Rust 세계에서 타의 추종을 불허합니다. 또한 이 코드를 다른 웹 프레임워크로 포팅하는 것은 쉽습니다.

채팅은 긴 폴링에서 가장 많은 이점을 얻는 교과서 응용 프로그램이므로 간단한 채팅 서버를 구현합니다.

이 구현을 효율적으로 만드는 3가지 요령이 있으니 주의를 기울이세요 ;)

채팅 서비스

채팅 서비스는 모든 비즈니스 로직을 캡슐화하는 개체입니다. 예제를 간단하게 유지하기 위해 데이터베이스 호출만 수행합니다.

첫 번째 요령은 다음과 같습니다. 메시지 순서 지정을 활성화하기 위해 UUIDv4 를 사용하지 않습니다. 대신 UUID로 변환하는 ULID을 사용하므로 직렬화/역직렬화하는 데 문제가 없습니다: Uuid = Ulid::new().into()
chat.rs

impl ChatService {
    pub fn new(db: DB) -> Self {
        ChatService { db }
    }

    pub async fn create_message(&self, body: String) -> Result<Message, Error> {
        if body.len() > 10_000 {
            return Err(Error::InvalidArgument("Message is too large".to_string()));
        }

        let created_at = chrono::Utc::now();
        let id: Uuid = Ulid::new().into();

        let query = "INSERT INTO messages
            (id, created_at, body)
            VALUES ($1, $2, $3)";

        sqlx::query(query)
            .bind(id)
            .bind(created_at)
            .bind(&body)
            .execute(&self.db)
            .await?;

        Ok(Message {
            id,
            created_at,
            body,
        })
    }

다음은 두 번째 트릭입니다. "0"UUID( after.unwrap_or(Uuid::nil()) )를 반환하는 00000000-0000-0000-0000-000000000000 에 주목하십시오. WHERE id > $1를 사용하면 after가 None인 경우 모든 메시지를 반환할 수 있습니다.

예를 들어 클라이언트의 전체 상태를 다시 수화하는 데 유용합니다.

    pub async fn find_messages(&self, after: Option<Uuid>) -> Result<Vec<Message>, Error> {
        let query = "SELECT *
            FROM messages
            WHERE id > $1";

        let messages: Vec<Message> = sqlx::query_as::<_, Message>(query)
            .bind(after.unwrap_or(Uuid::nil()))
            .fetch_all(&self.db)
            .await?;

        Ok(messages)
    }
}

웹 서버

다음으로 웹 서버를 실행하기 위한 상용구입니다.
.layer(AddExtensionLayer::new(ctx)) 덕분에 ServerContext가 모든 경로에 주입되어 ChatService의 메서드를 호출할 수 있습니다.

struct ServerContext {
    chat_service: chat::ChatService,
}

#[tokio::main]
async fn main() -> Result<(), anyhow::Error> {
    std::env::set_var("RUST_LOG", "rust_long_polling=info");
    env_logger::init();

    let database_url = std::env::var("DATABASE_URL")
        .map_err(|_| Error::BadConfig("DATABASE_URL env var is missing".to_string()))?;

    let db = db::connect(&database_url).await?;
    db::migrate(&db).await?;

    let chat_service = chat::ChatService::new(db);
    let ctx = Arc::new(ServerContext::new(chat_service));

    let app = Router::new()
        .route(
            "/messages",
            get(handler_find_messages).post(handler_create_message),
        )
        .or(handler_404.into_service())
        .layer(AddExtensionLayer::new(ctx));

    log::info!("Starting server on 0.0.0.0:8080");
    axum::Server::bind(
        &"0.0.0.0:8080"
            .parse()
            .expect("parsing server's bind address"),
    )
    .serve(app.into_make_service())
    .await
    .expect("running server");

    Ok(())
}

긴 폴링

마지막으로 세 번째 트릭: 긴 폴링은 tokio::time::sleep 를 사용하는 간단한 루프입니다.
tokio::time::sleep 를 사용하면 대기 중인 활성 연결에서 리소스를 거의 사용하지 않습니다.

새 데이터가 발견되면 즉시 새 데이터로 반환합니다. 그렇지 않으면 1초 더 기다립니다.

10초 후에 빈 데이터를 반환합니다.

main.rs

async fn handler_find_messages(
    Extension(ctx): Extension<Arc<ServerContext>>,
    query_params: Query<FindMessagesQueryParameters>,
) -> Result<Json<Vec<Message>>, Error> {
    let sleep_for = Duration::from_secs(1);

    // long polling: 10 secs
    for _ in 0..10u64 {
        let messages = ctx.chat_service.find_messages(query_params.after).await?;
        if messages.len() != 0 {
            return Ok(messages.into());
        }

        tokio::time::sleep(sleep_for).await;
    }

    // return an empty response
    Ok(Vec::new().into())
}

코드는 GitHub에 있습니다.

평소와 같이 GitHub에서 코드를 찾을 수 있습니다: github.com/skerkour/kerkour.com (저장소에 별표를 표시하는 것을 잊지 마세요 🙏).

Rust에서 고급 웹 애플리케이션(예: WebAssembly 프론트엔드 및 JSON API 백엔드)을 만드는 방법을 배우고 싶습니까? 제 책을 보세요: Black Hat Rust .