Add poem/subscription-redis example

Cargo.toml

@@ -7,6 +7,7 @@ members = [
 
     "poem/starwars",
     "poem/subscription",
+    "poem/subscription-redis",
     "poem/token-from-header",
 
     "actix-web/token-from-header",

poem/starwars/Cargo.toml

@@ -8,4 +8,4 @@ async-graphql = { path = "../../.." }
 async-graphql-poem = { path = "../../../integrations/poem" }
 tokio = { version = "1.8", features = ["macros", "rt-multi-thread"] }
 starwars = { path = "../../models/starwars" }
-poem = "1.2.2"
+poem = "1.2.25"

poem/subscription-redis/Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "subscription-redis"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+async-graphql = { path = "../../.." }
+async-graphql-poem = { path = "../../../integrations/poem" }
+tokio = { version = "1.8", features = ["macros", "rt-multi-thread"] }
+poem = { version = "1.2.25", features = ["websocket"] }
+redis = { version = "0.21.4", features = ["aio", "tokio-comp"] }
+futures-util = "0.3.19"

poem/subscription-redis/src/main.rs

@@ -0,0 +1,72 @@
+use async_graphql::http::{playground_source, GraphQLPlaygroundConfig};
+use async_graphql::{Context, Object, Result, Schema, Subscription};
+use async_graphql_poem::{GraphQL, GraphQLSubscription};
+use futures_util::{Stream, StreamExt};
+use poem::listener::TcpListener;
+use poem::web::Html;
+use poem::{get, handler, IntoResponse, Route, Server};
+use redis::{AsyncCommands, Client};
+
+struct QueryRoot;
+
+#[Object]
+impl QueryRoot {
+    async fn version(&self) -> &'static str {
+        std::env!("CARGO_PKG_VERSION")
+    }
+}
+
+struct MutationRoot;
+
+#[Object]
+impl MutationRoot {
+    async fn publish(&self, ctx: &Context<'_>, value: String) -> Result<bool> {
+        let client = ctx.data_unchecked::<Client>();
+        let mut conn = client.get_async_connection().await?;
+        conn.publish("values", value).await?;
+        Ok(true)
+    }
+}
+
+struct SubscriptionRoot;
+
+#[Subscription]
+impl SubscriptionRoot {
+    async fn values(&self, ctx: &Context<'_>) -> Result<impl Stream<Item = String>> {
+        let client = ctx.data_unchecked::<Client>();
+        let mut conn = client.get_async_connection().await?.into_pubsub();
+        conn.subscribe("values").await?;
+        Ok(conn
+            .into_on_message()
+            .filter_map(|msg| async move { msg.get_payload().ok() }))
+    }
+}
+
+#[handler]
+async fn graphql_playground() -> impl IntoResponse {
+    Html(playground_source(
+        GraphQLPlaygroundConfig::new("/").subscription_endpoint("/ws"),
+    ))
+}
+
+#[tokio::main]
+async fn main() {
+    let client = Client::open("redis://127.0.0.1/").unwrap();
+
+    let schema = Schema::build(QueryRoot, MutationRoot, SubscriptionRoot)
+        .data(client)
+        .finish();
+
+    let app = Route::new()
+        .at(
+            "/",
+            get(graphql_playground).post(GraphQL::new(schema.clone())),
+        )
+        .at("/ws", get(GraphQLSubscription::new(schema)));
+
+    println!("Playground: http://localhost:8000");
+    Server::new(TcpListener::bind("0.0.0.0:8000"))
+        .run(app)
+        .await
+        .unwrap();
+}

poem/subscription/Cargo.toml

@@ -8,4 +8,4 @@ async-graphql = { path = "../../.." }
 async-graphql-poem = { path = "../../../integrations/poem" }
 tokio = { version = "1.8", features = ["macros", "rt-multi-thread"] }
 books = { path = "../../models/books" }
-poem = { version = "1.2.2", features = ["websocket"] }
+poem = { version = "1.2.25", features = ["websocket"] }

poem/token-from-header/Cargo.toml

@@ -7,5 +7,5 @@ edition = "2021"
 async-graphql = { path = "../../.." }
 async-graphql-poem = { path = "../../../integrations/poem" }
 token = { path = "../../models/token" }
-poem = { version = "1.2.2", features = ["websocket"] }
+poem = { version = "1.2.25", features = ["websocket"] }
 tokio = { version = "1.8", features = ["macros", "rt-multi-thread"] }