This is a port of the Rocket Todo example into actix-web. Except this uses PostgreSQL instead of SQLite.

Is this something you would like to include in these examples?

It integrates:

• Diesel with PostgreSQL
• Tera templates
• Logging
• Serving CSS static files
• Cookie sessions
• Differentiating between PUT and DELETE using a hidden _method form field
• Flash messages via the session cookie

Note:

• This is the first code I've written using actix, so it might not be particularly "idiomatic". Please let me know if there are better ways to do things 🙂
• I deliberately used PostgreSQL instead of SQLite because I wanted to create an example using an RDBMS I'm more likely to use in production. Would we need to port to SQLite for inclusion in these examples?

Todo

• [x] add this sub-project to the Travis file
• [x] rebase if/when #40 is merged
• [x] Rework DBExecutor
• [x] Try to rewrite the HTTP-method reinterpretation code as middleware
• [x] Use templates for 400/500 errors

This PR Hopefully creates a minimal autobahn websocket server example from which to start testing and addressing issues for https://github.com/actix/actix-web/issues/1006.

I assume this example needs to be tweaked slightly to make a reasonable benchmark. From there I can run the autobahn test suite and document all the failures. Then we can pick them off as we see fit.

Related to https://github.com/actix/examples/issues/61

This is a non-functional example of what I think is an important use case: Being able to post to an actor and then return a json response. It would be great to have this reviewed and corrected to functionally work.

I run the example, It is very good.bug how I get app_state from &executor? I am a beginner, not very understanding of the operation mechanism, will there be more explanation?

rusqlite have some breaking changes and new features, this PR

1. updates rusqlite in async_db and r2d2 example
2. adds a note about the use of web::block
3. deduplicates repetitive code
4. uses the types with Iterator::collect to be idiomatic

Server:

[mkpankov@mkpankov-lenovo-x240]~/work/actix/examples/multipart% cargo run 
    Finished dev [unoptimized + debuginfo] target(s) in 0.62s
     Running `/home/mkpankov/work/actix/examples/target/debug/multipart`
[2019-08-10T14:18:52Z INFO  actix_server::builder] Starting 4 workers
[2019-08-10T14:18:52Z INFO  actix_server::builder] Starting server on 127.0.0.1:8080
1
[2019-08-10T14:18:56Z INFO  actix_web::middleware::logger] 127.0.0.1:36536 "POST / HTTP/1.1" 200 6 "-" "Python/3.7 aiohttp/3.5.4" 0.001850
1
failed: Multipart boundary is not found
[2019-08-10T14:18:56Z INFO  actix_web::middleware::logger] 127.0.0.1:36540 "POST / HTTP/1.1" 500 31 "-" "Python/3.7 aiohttp/3.5.4" 0.001922

Client:

[mkpankov@mkpankov-lenovo-x240]~/work/actix/examples/multipart% python3 client.py
Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x7fc9a4535cc0>
Unclosed connector
connections: ['[(<aiohttp.client_proto.ResponseHandler object at 0x7fc9a41f2588>, 1789.448346384)]']
connector: <aiohttp.connector.TCPConnector object at 0x7fc9a4535a58>
<ClientResponse(http://localhost:8080/) [200 OK]>
<CIMultiDictProxy('Content-Length': '6', 'Content-Type': 'application/json', 'Date': 'Sat, 10 Aug 2019 14:19:36 GMT')>

<ClientResponse(http://localhost:8080/) [500 Internal Server Error]>
<CIMultiDictProxy('Content-Length': '31', 'Content-Type': 'text/plain', 'Date': 'Sat, 10 Aug 2019 14:19:36 GMT')>

Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x7fc9a41d3160>
Unclosed connector
connections: ['[(<aiohttp.client_proto.ResponseHandler object at 0x7fc9a41f24c8>, 1789.46075974)]']
connector: <aiohttp.connector.TCPConnector object at 0x7fc9a41d3208>
Traceback (most recent call last):
  File "client.py", line 36, in <module>
    loop.run_until_complete(req2())
  File "/usr/lib64/python3.7/asyncio/base_events.py", line 584, in run_until_complete
    return future.result()
  File "client.py", line 31, in req2
    assert 200 == resp.status
AssertionError

• added commands:

/list-clients
/whoami

• correctly removing client from room when he d/c or joins another room
• removing code that has client rejoin room on every msg he sends
• updating web client for feats
• other refactors for clarity
• more logging

I don't know how to split these two commits.

The first one is about pool.get() block the current thread, i moved it into web::block block.

The second commit is a compatibility for windows.

Actix Actor Message Passsing

This is my example repository where I am attempting to understand how to compose the output of various actors into something that actix_web can use as a response.

The example shows how to construct a chain of actors (sync) in an asynchronous way and provides an ergonomic way of doing so. The important things to not are the usage of a single error struct struct SystemError and the flatten() method to coalesce the future response types.

Hopefully this can be useful and I hope to get some feedback. Been back using rust for one week, so I'm ready for it all to be considered awful 😄

Hi, is it presently possible to use actix-web for serverless?

In other words: the application is invoked to render a single endpoint before terminating.

So, rather than the usual

fn main() {
    let body = server::new(
        || App::new()
            .route("/{id}/{name}/index.html", http::Method::GET, index))
        .bind("127.0.0.1:8080").unwrap()
        .run();
}

I'm looking for something like

fn main() {
    let args = std::env::args().into_iter();
    let method: http::Method = args.next().unwrap_or("GET").parse();
    let path = args.next().unwrap_or("/");

    let body = server::new(
        || App::new()
            .route("/{id}/{name}/index.html", http::Method::GET, index))
        .invoke(method, path)
        .expect("Failed to render path.");

    println!("{}", Responder::new(body));
}

Request with payload (Middleware activated)

Request with same payload (Middleware deactivated)

How I set the middleware:

The middleware struct & implementation. Copied straight from https://github.com/actix/examples/blob/master/middleware/middleware/src/read_request_body.rs

use std::{future::{ready, Ready}, rc::Rc, io::Read};

use actix_web::{
    dev::{self, Service, ServiceRequest, ServiceResponse, Transform},
    Error, HttpMessage, web::BytesMut,
};
use futures_util::{future::LocalBoxFuture, StreamExt, Stream};
// There are two steps in middleware processing.
// 1. Middleware initialization, middleware factory gets called with
//    next service in chain as parameter.
// 2. Middleware's call method gets called with normal request.
pub struct BodyLogger;

// Middleware factory is `Transform` trait from actix-service crate
// `S` - type of the next service
// `B` - type of response's body
impl<S: 'static, B> Transform<S, ServiceRequest> for BodyLogger
where
    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
    S::Future: 'static,
    B: 'static,
{
    type Response = ServiceResponse<B>;
    type Error = Error;
    type InitError = ();
    type Transform = BodyLoggerMiddleware<S>;
    type Future = Ready<Result<Self::Transform, Self::InitError>>;

    fn new_transform(&self, service: S) -> Self::Future {
        ready(Ok(BodyLoggerMiddleware { service: Rc::new(service) }))
    }
}

pub struct BodyLoggerMiddleware<S> {
    service: Rc<S>,
}

impl<S, B> Service<ServiceRequest> for BodyLoggerMiddleware<S>
where
    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error> + 'static,
    S::Future: 'static,
    B: 'static,
{
    type Response = ServiceResponse<B>;
    type Error = Error;
    type Future = LocalBoxFuture<'static, Result<Self::Response, Self::Error>>;

    dev::forward_ready!(service);

    fn call(&self, mut req: ServiceRequest) -> Self::Future {
        let svc = self.service.clone();
        Box::pin(async move {

            let mut body = BytesMut::new();
            let mut stream = req.take_payload();
            while let Some(chunk) = stream.next().await {
                body.extend_from_slice(&chunk?);
            }

            println!("request body: {body:?}");
            let res = svc.call(req).await?;

            println!("response: {:?}", res.headers());
            Ok(res)
        })
    }
}

I think these might be two issues in one:

1. There's no example (except maybe the SSE one) that has streaming responses. Which is probably because it's hard to come up with a good example for streaming responses.
2. I haven't found good documentation on how to "pipe" an AsyncWrite into a Stream<Item=Bytes(Mut)> (maybe something for actix-web-lab).

Note: Also see discussion on actix-web Discord: https://discord.com/channels/771444961383153695/771447545154371646/1009110232473014383

The proposed example is one that streams files from a directory (files) as a zip file, i.e. it dynamically creates the zip file.

For this I'm using async_zip which exposes an API that requires the user to pass in an AsyncWrite (ZipFileWriter::new). To "pipe" the AsyncWrite to a Stream, I'm using a DuplexStream and the BytesCodec.

use actix_web::{get, http, App, HttpResponse, HttpServer, Responder};
use async_zip::write::{EntryOptions, ZipFileWriter};
use futures::stream::TryStreamExt;
use std::io;
use tokio::io::AsyncWrite;
use tokio_util::codec;

#[get("/")]
async fn index() -> impl Responder {
    let (to_write, to_read) = tokio::io::duplex(2048);
    tokio::spawn(async move {
        let mut zipper = async_zip::write::ZipFileWriter::new(to_write);
        if let Err(e) = read_dir(&mut zipper).await {
            // TODO: do something
            eprintln!("Failed to write files from directory to zip: {e}")
        }
        if let Err(e) = zipper.close().await {
            // TODO: do something
            eprintln!("Failed to close zipper: {e}")
        }
    });

    let stream = codec::FramedRead::new(to_read, codec::BytesCodec::new()).map_ok(|b| b.freeze());
    HttpResponse::Ok()
        .append_header((
            http::header::CONTENT_DISPOSITION,
            r#"attachment; filename="folder.zip""#,
        ))
        // not sure if this is really necessary,
        // but we're already sending compressed data,
        // so make sure other middleware won't compress this again
        .append_header((http::header::CONTENT_ENCODING, "identity"))
        .streaming(stream)
}

async fn read_dir<W>(zipper: &mut ZipFileWriter<W>) -> Result<(), io::Error>
where
    W: AsyncWrite + Unpin,
{
    let mut dir = tokio::fs::read_dir("files").await?;
    while let Ok(Some(entry)) = dir.next_entry().await {
        if !entry.metadata().await.map(|m| m.is_file()).unwrap_or(false) {
            continue;
        }
        let mut file = match tokio::fs::OpenOptions::new()
            .read(true)
            .open(entry.path())
            .await
        {
            Ok(f) => f,
            Err(_) => continue, // we can't read the file
        };
        let filename = match entry.file_name().into_string() {
            Ok(s) => s,
            Err(_) => continue, // the file has a non UTF-8 name
        };

        let mut entry = zipper
            .write_entry_stream(EntryOptions::new(filename, async_zip::Compression::Deflate))
            .await
            .map_err(zip_to_io_err)?;
        tokio::io::copy(&mut file, &mut entry).await?;
        entry.close().await.map_err(zip_to_io_err)?;
    }
    Ok(())
}

fn zip_to_io_err(e: async_zip::error::ZipError) -> io::Error {
    io::Error::new(io::ErrorKind::Other, e)
}

#[actix_web::main]
async fn main() -> io::Result<()> {
    HttpServer::new(move || App::new().service(index))
        .bind(("127.0.0.1", 8080))?
        .run()
        .await
}

As I explained on Discord, using a DuplexStream is probably overkill, since it's supposed to be used bi-directional (see example in tokio docs), so I tried to extract the internal Pipe used by the tokio implementation and made a pipe specifically for (buffered) piping of AsyncWrite to a Stream<BytesMut>. I'm not sure if this should be included in actix-web-lab as a utility when dealing with AsyncWrite (or maybe in some other crate?).

use std::{sync::{Arc, Mutex, MutexGuard}, task::{self,Waker, Poll}, pin::Pin};

use bytes::{BytesMut, Buf};
use futures::Stream;
use tokio::io::AsyncWrite;

/// The `AsyncWrite` half of an [`async_pipe`]
pub struct AsyncPipeWriter(Arc<Mutex<Pipe>>);
/// The `Stream` half of an [`async_pipe`]
pub struct AsyncPipeReader(Arc<Mutex<Pipe>>);

/// Creates buffered pipe that pipes writes from an `AsyncWrite` to a `Stream<Item=BytesMut>`.
/// 
/// `max_buf_size` is the maximum amount of bytes that can be written to the pipe's internal buffer
/// before further writes return `Poll::Pending`.
pub fn async_pipe(max_buf_size: usize) -> (AsyncPipeWriter, AsyncPipeReader) {
    let pipe = Arc::new(Mutex::new(Pipe::new(max_buf_size)));
    (AsyncPipeWriter(pipe.clone()), AsyncPipeReader(pipe))
}

/// A unidirectional IO over a piece of memory.
///
/// Data can be written to the pipe, and reading will return that data.
/// 
/// [tokio's](https://github.com/tokio-rs/tokio/blob/de81985762a242c77361a6ab9de198372ca85987/tokio/src/io/util/mem.rs#L54-L76)
/// internal representation of a pipe for a `tokio::io::DuplexStream`.
#[derive(Debug)]
struct Pipe {
    /// The buffer storing the bytes written, also read from.
    ///
    /// Using a `BytesMut` because it has efficient `Buf` and `BufMut`
    /// functionality already.
    buffer: BytesMut,
    /// Determines if the write side has been closed.
    is_closed: bool,
    /// The maximum amount of bytes that can be written before returning
    /// `Poll::Pending`.
    max_buf_size: usize,
    /// If the `read` side has been polled and is pending, this is the waker
    /// for that parked task.
    read_waker: Option<Waker>,
    /// If the `write` side has filled the `max_buf_size` and returned
    /// `Poll::Pending`, this is the waker for that parked task.
    write_waker: Option<Waker>,
}

impl Pipe {
    fn new(max_buf_size: usize) -> Self {
        Pipe {
            buffer: BytesMut::new(),
            is_closed: false,
            max_buf_size,
            read_waker: None,
            write_waker: None,
        }
    }

    fn close_write(&mut self) {
        self.is_closed = true;
        // needs to notify any readers that no more data will come
        if let Some(waker) = self.read_waker.take() {
            waker.wake();
        }
    }

    fn close_read(&mut self) {
        self.is_closed = true;
        // needs to notify any writers that they have to abort
        if let Some(waker) = self.write_waker.take() {
            waker.wake();
        }
    }

    fn poll_read_internal(
        mut self: Pin<&mut Self>,
        cx: &mut task::Context<'_>
    ) -> Poll<Option<BytesMut>> {
        if self.buffer.has_remaining() {
            let bytes = std::mem::take(&mut self.buffer);
            if let Some(waker) = self.write_waker.take() {
                waker.wake();
            }
            Poll::Ready(Some(bytes))
        } else if self.is_closed {
            Poll::Ready(None)
        } else {
            self.read_waker = Some(cx.waker().clone());
            Poll::Pending
        }
    }

    fn poll_write_internal(
        mut self: Pin<&mut Self>,
        cx: &mut task::Context<'_>,
        buf: &[u8],
    ) -> Poll<std::io::Result<usize>> {
        if self.is_closed {
            return Poll::Ready(Err(std::io::ErrorKind::BrokenPipe.into()));
        }
        let avail = self.max_buf_size - self.buffer.len();
        if avail == 0 {
            self.write_waker = Some(cx.waker().clone());
            return Poll::Pending;
        }

        let len = buf.len().min(avail);
        self.buffer.extend_from_slice(&buf[..len]);
        if let Some(waker) = self.read_waker.take() {
            waker.wake();
        }
        Poll::Ready(Ok(len))
    }
}

impl AsyncWrite for Pipe {
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut task::Context<'_>,
        buf: &[u8],
    ) -> Poll<std::io::Result<usize>> {
        self.poll_write_internal(cx, buf)
    }

    fn poll_flush(self: Pin<&mut Self>, _: &mut task::Context<'_>) -> Poll<std::io::Result<()>> {
        Poll::Ready(Ok(()))
    }

    fn poll_shutdown(
        mut self: Pin<&mut Self>,
        _: &mut task::Context<'_>,
    ) -> Poll<std::io::Result<()>> {
        self.close_write();
        Poll::Ready(Ok(()))
    }
}

impl AsyncWrite for AsyncPipeWriter {
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut task::Context<'_>,
        buf: &[u8],
    ) -> Poll<std::io::Result<usize>> {
        Pin::new(&mut *always_lock(&self.0)).poll_write(cx, buf)
    }

    fn poll_flush(
        self: Pin<&mut Self>,
        cx: &mut task::Context<'_>,
    ) -> Poll<std::io::Result<()>> {
        Pin::new(&mut *always_lock(&self.0)).poll_flush(cx)
    }

    fn poll_shutdown(
        self: Pin<&mut Self>,
        cx: &mut task::Context<'_>,
    ) -> Poll<std::io::Result<()>> {
        Pin::new(&mut *always_lock(&self.0)).poll_shutdown(cx)
    }
}

impl Stream for Pipe {
    type Item = BytesMut;

    fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
        self.poll_read_internal(cx)
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        let remaining = self.buffer.remaining();
        (remaining, Some(remaining))
    }
}

impl Stream for AsyncPipeReader {
    type Item = BytesMut;

    fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
        Pin::new(&mut *always_lock(&self.0)).poll_next(cx)
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        always_lock(&self.0).size_hint()
    }
}

impl Drop for AsyncPipeWriter {
    fn drop(&mut self) {
        // notify the other side of the closure
        always_lock(&self.0).close_write();
    }
}

impl Drop for AsyncPipeReader {
    fn drop(&mut self) {
        // notify the other side of the closure
        always_lock(&self.0).close_read();
    }
}

#[inline]
fn always_lock<T>(mtx: &Mutex<T>) -> MutexGuard<T> {
    match mtx.lock() {
        Ok(g) => g,
        Err(e) => e.into_inner(),
    }
}

#[get("/")]
async fn index() -> impl Responder {
    let (to_write, stream) = async_pipe(2048);
    tokio::spawn(async move {
        let mut zipper = async_zip::write::ZipFileWriter::new(to_write);
        if let Err(e) = read_dir(&mut zipper).await {
            // TODO: do something
            eprintln!("Failed to write files from directory to zip: {e}")
        }
        if let Err(e) = zipper.close().await {
            // TODO: do something
            eprintln!("Failed to close zipper: {e}")
        }
    });
    HttpResponse::Ok()
        .append_header((
            http::header::CONTENT_DISPOSITION,
            r#"attachment; filename="folder.zip""#,
        ))
        // not sure if this is really necessary,
        // but we're already sending compressed data,
        // so make sure other middleware won't compress this again
        .append_header((http::header::CONTENT_ENCODING, "identity"))
        .streaming(stream.map(|b| Ok::<_, io::Error>(b.freeze())))
}

[package]
name = "actix-zippy"
version = "0.1.0"
edition = "2021"

[dependencies]
actix-web = "4.1.0"
async_zip = "0.0.8"
bytes = "1.2.1"
futures = "0.3.23"
tokio = { version = "1.20.1", features = ["io-util", "fs"] }
tokio-stream = "0.1.9"
tokio-util = { version = "0.7.3", features = ["codec"] }

The example here: https://github.com/actix/examples/blob/master/databases/redis/src/main.rs

could be updated with the r2d2 example of pooled Redis client connections from here: https://github.com/redis-rs/redis-rs/pull/388/files

The following code worked for me:

use actix_web::{get, web, App, HttpServer, Responder, Result};
use redis::Commands;

#[get("/simple/url/{id}")]
pub async fn example(
    id: web::Path<String>,
    redis_pool: web::Data<r2d2::Pool<redis::Client>>
) -> Result<impl Responder> {
    let mut conn: r2d2::PooledConnection<redis::Client> = redis_pool.get().unwrap();

    // use the connection as usual
    let _: () = conn.set("KEY", "VALUE").unwrap();
    let val: String = conn.get("KEY").unwrap();
    println!("{}", val);

    Ok(web::Json(val))
}

#[actix_web::main] // or #[tokio::main]
async fn main() -> std::io::Result<()> {
    let client = redis::Client::open("redis://127.0.0.1/").unwrap();
    let pool: r2d2::Pool<redis::Client> = r2d2::Pool::builder()
        .max_size(15)
        .build(client)
        .unwrap_or_else(|e| panic!("Error building redis pool: {}", e));
    HttpServer::new(move || {
        App::new()
            .app_data(web::Data::new(pool.clone()))
            .route("/hello", web::get().to(|| async { "Hello World!" }))
            .service(example)
    })
        .bind(("127.0.0.1", 8080))?
        .run()
        .await
}

I'm learning about how to implement web sockets with actix and I was looking at the example linked from here:

I understand the example, it's clear and I understand everything, but I don't get why is this python file there, I got 404 everytime that I run it :/ https://github.com/actix/examples/blob/master/websockets/chat/client.py

This issue is to express the wish that actix-web get a mysql example

Like #397 and #405 is this issue tagged new-example.

I'm aware of mariadb, reason for having mysql in the issue title is that the protocol is named mysql.

And I hope that this issue attracts messages like "At some_URL is the combo actix-web and mysql being used"