2020-03-17 09:26:59 +00:00
|
|
|
use actix::{
|
2020-09-17 18:22:54 +00:00
|
|
|
Actor, ActorContext, ActorFuture, ActorStream, AsyncContext, ContextFutureSpawner,
|
|
|
|
StreamHandler, WrapFuture, WrapStream,
|
2020-03-17 09:26:59 +00:00
|
|
|
};
|
2020-09-17 18:22:54 +00:00
|
|
|
use actix_http::ws;
|
2020-03-17 09:26:59 +00:00
|
|
|
use actix_web_actors::ws::{Message, ProtocolError, WebsocketContext};
|
2020-09-17 18:22:54 +00:00
|
|
|
use async_graphql::http::WebSocket;
|
2020-09-12 09:29:52 +00:00
|
|
|
use async_graphql::{resolver_utils::ObjectType, Data, FieldResult, Schema, SubscriptionType};
|
2020-09-17 18:22:54 +00:00
|
|
|
use futures::channel::mpsc;
|
|
|
|
use futures::SinkExt;
|
2020-03-19 09:20:12 +00:00
|
|
|
use std::time::{Duration, Instant};
|
2020-03-17 09:26:59 +00:00
|
|
|
|
2020-04-07 06:30:46 +00:00
|
|
|
const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
|
|
|
|
const CLIENT_TIMEOUT: Duration = Duration::from_secs(10);
|
|
|
|
|
2020-04-14 01:53:17 +00:00
|
|
|
/// Actor for subscription via websocket
|
|
|
|
pub struct WSSubscription<Query, Mutation, Subscription> {
|
2020-09-17 18:22:54 +00:00
|
|
|
schema: Option<Schema<Query, Mutation, Subscription>>,
|
|
|
|
last_heartbeat: Instant,
|
|
|
|
messages: Option<mpsc::UnboundedSender<Vec<u8>>>,
|
|
|
|
initializer: Option<Box<dyn FnOnce(serde_json::Value) -> FieldResult<Data> + Send + Sync>>,
|
|
|
|
continuation: Vec<u8>,
|
2020-03-17 09:26:59 +00:00
|
|
|
}
|
|
|
|
|
2020-04-14 01:53:17 +00:00
|
|
|
impl<Query, Mutation, Subscription> WSSubscription<Query, Mutation, Subscription>
|
2020-03-17 09:26:59 +00:00
|
|
|
where
|
2020-03-19 09:20:12 +00:00
|
|
|
Query: ObjectType + Send + Sync + 'static,
|
|
|
|
Mutation: ObjectType + Send + Sync + 'static,
|
|
|
|
Subscription: SubscriptionType + Send + Sync + 'static,
|
2020-03-17 09:26:59 +00:00
|
|
|
{
|
2020-04-14 01:53:17 +00:00
|
|
|
/// Create an actor for subscription connection via websocket.
|
2020-09-17 18:22:54 +00:00
|
|
|
pub fn new(schema: Schema<Query, Mutation, Subscription>) -> Self {
|
2020-03-17 09:26:59 +00:00
|
|
|
Self {
|
2020-09-17 18:22:54 +00:00
|
|
|
schema: Some(schema),
|
|
|
|
last_heartbeat: Instant::now(),
|
|
|
|
messages: None,
|
2020-09-11 09:52:06 +00:00
|
|
|
initializer: None,
|
2020-09-17 18:22:54 +00:00
|
|
|
continuation: Vec::new(),
|
2020-04-23 07:30:12 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Set a context data initialization function.
|
2020-09-11 09:52:06 +00:00
|
|
|
pub fn initializer<F>(self, f: F) -> Self
|
2020-04-23 07:30:12 +00:00
|
|
|
where
|
2020-09-17 18:22:54 +00:00
|
|
|
F: FnOnce(serde_json::Value) -> FieldResult<Data> + Send + Sync + 'static,
|
2020-04-23 07:30:12 +00:00
|
|
|
{
|
|
|
|
Self {
|
2020-09-11 09:52:06 +00:00
|
|
|
initializer: Some(Box::new(f)),
|
2020-04-23 07:30:12 +00:00
|
|
|
..self
|
2020-03-17 09:26:59 +00:00
|
|
|
}
|
|
|
|
}
|
2020-03-19 09:20:12 +00:00
|
|
|
|
2020-09-17 18:22:54 +00:00
|
|
|
fn send_heartbeats(&self, ctx: &mut WebsocketContext<Self>) {
|
2020-04-07 06:30:46 +00:00
|
|
|
ctx.run_interval(HEARTBEAT_INTERVAL, |act, ctx| {
|
2020-09-17 18:22:54 +00:00
|
|
|
if Instant::now().duration_since(act.last_heartbeat) > CLIENT_TIMEOUT {
|
2020-03-19 09:20:12 +00:00
|
|
|
ctx.stop();
|
|
|
|
}
|
2020-04-07 06:30:46 +00:00
|
|
|
ctx.ping(b"");
|
2020-03-19 09:20:12 +00:00
|
|
|
});
|
|
|
|
}
|
2020-03-17 09:26:59 +00:00
|
|
|
}
|
|
|
|
|
2020-04-14 01:53:17 +00:00
|
|
|
impl<Query, Mutation, Subscription> Actor for WSSubscription<Query, Mutation, Subscription>
|
2020-03-17 09:26:59 +00:00
|
|
|
where
|
2020-03-19 09:20:12 +00:00
|
|
|
Query: ObjectType + Sync + Send + 'static,
|
|
|
|
Mutation: ObjectType + Sync + Send + 'static,
|
|
|
|
Subscription: SubscriptionType + Send + Sync + 'static,
|
2020-03-17 09:26:59 +00:00
|
|
|
{
|
|
|
|
type Context = WebsocketContext<Self>;
|
|
|
|
|
|
|
|
fn started(&mut self, ctx: &mut Self::Context) {
|
2020-09-17 18:22:54 +00:00
|
|
|
self.send_heartbeats(ctx);
|
|
|
|
|
|
|
|
let (tx, rx) = mpsc::unbounded();
|
|
|
|
|
|
|
|
WebSocket::with_data(self.schema.take().unwrap(), rx, self.initializer.take())
|
|
|
|
.into_actor(self)
|
|
|
|
.map(|response, _act, ctx| {
|
|
|
|
ctx.text(response);
|
|
|
|
})
|
|
|
|
.finish()
|
|
|
|
.spawn(ctx);
|
|
|
|
|
|
|
|
self.messages = Some(tx);
|
2020-03-17 09:26:59 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl<Query, Mutation, Subscription> StreamHandler<Result<Message, ProtocolError>>
|
2020-04-14 01:53:17 +00:00
|
|
|
for WSSubscription<Query, Mutation, Subscription>
|
2020-03-17 09:26:59 +00:00
|
|
|
where
|
2020-03-19 09:20:12 +00:00
|
|
|
Query: ObjectType + Sync + Send + 'static,
|
|
|
|
Mutation: ObjectType + Sync + Send + 'static,
|
|
|
|
Subscription: SubscriptionType + Send + Sync + 'static,
|
2020-03-17 09:26:59 +00:00
|
|
|
{
|
|
|
|
fn handle(&mut self, msg: Result<Message, ProtocolError>, ctx: &mut Self::Context) {
|
|
|
|
let msg = match msg {
|
|
|
|
Err(_) => {
|
|
|
|
ctx.stop();
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
Ok(msg) => msg,
|
|
|
|
};
|
|
|
|
|
2020-09-17 18:22:54 +00:00
|
|
|
let message = match msg {
|
2020-03-17 09:26:59 +00:00
|
|
|
Message::Ping(msg) => {
|
2020-09-17 18:22:54 +00:00
|
|
|
self.last_heartbeat = Instant::now();
|
2020-03-17 09:26:59 +00:00
|
|
|
ctx.pong(&msg);
|
2020-09-17 18:22:54 +00:00
|
|
|
None
|
2020-03-17 09:26:59 +00:00
|
|
|
}
|
|
|
|
Message::Pong(_) => {
|
2020-09-17 18:22:54 +00:00
|
|
|
self.last_heartbeat = Instant::now();
|
|
|
|
None
|
2020-03-17 09:26:59 +00:00
|
|
|
}
|
2020-09-17 18:22:54 +00:00
|
|
|
Message::Continuation(item) => match item {
|
|
|
|
ws::Item::FirstText(bytes) | ws::Item::FirstBinary(bytes) => {
|
|
|
|
self.continuation = bytes.to_vec();
|
|
|
|
None
|
2020-03-17 09:26:59 +00:00
|
|
|
}
|
2020-09-17 18:22:54 +00:00
|
|
|
ws::Item::Continue(bytes) => {
|
|
|
|
self.continuation.extend_from_slice(&bytes);
|
|
|
|
None
|
|
|
|
}
|
|
|
|
ws::Item::Last(bytes) => {
|
|
|
|
self.continuation.extend_from_slice(&bytes);
|
|
|
|
Some(std::mem::take(&mut self.continuation))
|
|
|
|
}
|
|
|
|
},
|
|
|
|
Message::Text(s) => Some(s.into_bytes()),
|
|
|
|
Message::Binary(bytes) => Some(bytes.to_vec()),
|
|
|
|
Message::Close(_) => {
|
2020-03-17 09:26:59 +00:00
|
|
|
ctx.stop();
|
2020-09-17 18:22:54 +00:00
|
|
|
None
|
2020-03-17 09:26:59 +00:00
|
|
|
}
|
2020-09-17 18:22:54 +00:00
|
|
|
Message::Nop => None,
|
|
|
|
};
|
2020-03-17 09:26:59 +00:00
|
|
|
|
2020-09-17 18:22:54 +00:00
|
|
|
if let Some(message) = message {
|
|
|
|
let mut sender = self.messages.as_ref().unwrap().clone();
|
|
|
|
|
|
|
|
async move { sender.send(message).await }
|
|
|
|
.into_actor(self)
|
|
|
|
.map(|res, _actor, ctx| match res {
|
|
|
|
Ok(()) => {}
|
|
|
|
Err(_) => ctx.stop(),
|
|
|
|
})
|
|
|
|
.spawn(ctx)
|
|
|
|
}
|
2020-03-17 09:26:59 +00:00
|
|
|
}
|
|
|
|
}
|