Fix internal context missing when there's no connection_init frame. #451
This commit is contained in:
parent
d280978ccb
commit
e991ffb814
@ -61,7 +61,7 @@ pin_project! {
|
|||||||
pub struct WebSocket<S, F, Query, Mutation, Subscription> {
|
pub struct WebSocket<S, F, Query, Mutation, Subscription> {
|
||||||
data_initializer: Option<F>,
|
data_initializer: Option<F>,
|
||||||
init_fut: Option<BoxFuture<'static, Result<Data>>>,
|
init_fut: Option<BoxFuture<'static, Result<Data>>>,
|
||||||
data: Arc<Data>,
|
data: Option<Arc<Data>>,
|
||||||
schema: Schema<Query, Mutation, Subscription>,
|
schema: Schema<Query, Mutation, Subscription>,
|
||||||
streams: HashMap<String, Pin<Box<dyn Stream<Item = Response> + Send>>>,
|
streams: HashMap<String, Pin<Box<dyn Stream<Item = Response> + Send>>>,
|
||||||
#[pin]
|
#[pin]
|
||||||
@ -105,7 +105,7 @@ impl<S, F, Query, Mutation, Subscription> WebSocket<S, F, Query, Mutation, Subsc
|
|||||||
Self {
|
Self {
|
||||||
data_initializer: Some(data_initializer),
|
data_initializer: Some(data_initializer),
|
||||||
init_fut: None,
|
init_fut: None,
|
||||||
data: Arc::default(),
|
data: None,
|
||||||
schema,
|
schema,
|
||||||
streams: HashMap::new(),
|
streams: HashMap::new(),
|
||||||
stream,
|
stream,
|
||||||
@ -174,13 +174,17 @@ where
|
|||||||
id,
|
id,
|
||||||
payload: request,
|
payload: request,
|
||||||
} => {
|
} => {
|
||||||
this.streams.insert(
|
if let Some(data) = this.data.clone() {
|
||||||
id,
|
this.streams.insert(
|
||||||
Box::pin(
|
id,
|
||||||
this.schema
|
Box::pin(this.schema.execute_stream_with_ctx_data(request, data)),
|
||||||
.execute_stream_with_ctx_data(request, Arc::clone(this.data)),
|
);
|
||||||
),
|
} else {
|
||||||
);
|
return Poll::Ready(Some(WsMessage::Close(
|
||||||
|
1011,
|
||||||
|
"The handshake is not completed.".to_string(),
|
||||||
|
)));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
ClientMessage::Stop { id } => {
|
ClientMessage::Stop { id } => {
|
||||||
if this.streams.remove(id).is_some() {
|
if this.streams.remove(id).is_some() {
|
||||||
@ -202,7 +206,7 @@ where
|
|||||||
*this.init_fut = None;
|
*this.init_fut = None;
|
||||||
return match res {
|
return match res {
|
||||||
Ok(data) => {
|
Ok(data) => {
|
||||||
*this.data = Arc::new(data);
|
*this.data = Some(Arc::new(data));
|
||||||
Poll::Ready(Some(WsMessage::Text(
|
Poll::Ready(Some(WsMessage::Text(
|
||||||
serde_json::to_string(&ServerMessage::ConnectionAck).unwrap(),
|
serde_json::to_string(&ServerMessage::ConnectionAck).unwrap(),
|
||||||
)))
|
)))
|
||||||
|
@ -430,3 +430,37 @@ pub async fn test_query_over_websocket() {
|
|||||||
serde_json::from_str(&stream.next().await.unwrap().unwrap_text()).unwrap()
|
serde_json::from_str(&stream.next().await.unwrap().unwrap_text()).unwrap()
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
pub async fn test_start_before_connection_init() {
|
||||||
|
struct QueryRoot;
|
||||||
|
|
||||||
|
#[Object]
|
||||||
|
impl QueryRoot {
|
||||||
|
async fn value(&self) -> i32 {
|
||||||
|
999
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let schema = Schema::new(QueryRoot, EmptyMutation, EmptySubscription);
|
||||||
|
let (mut tx, rx) = mpsc::unbounded();
|
||||||
|
let mut stream = http::WebSocket::new(schema, rx, WebSocketProtocols::GraphQLWS);
|
||||||
|
|
||||||
|
tx.send(
|
||||||
|
serde_json::to_string(&value!({
|
||||||
|
"type": "start",
|
||||||
|
"id": "1",
|
||||||
|
"payload": {
|
||||||
|
"query": "query { value }"
|
||||||
|
},
|
||||||
|
}))
|
||||||
|
.unwrap(),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
stream.next().await.unwrap().unwrap_close(),
|
||||||
|
(1011, "The handshake is not completed.".to_string())
|
||||||
|
);
|
||||||
|
}
|
||||||
|
@ -391,3 +391,37 @@ pub async fn test_query_over_websocket() {
|
|||||||
serde_json::from_str(&stream.next().await.unwrap().unwrap_text()).unwrap()
|
serde_json::from_str(&stream.next().await.unwrap().unwrap_text()).unwrap()
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
pub async fn test_start_before_connection_init() {
|
||||||
|
struct QueryRoot;
|
||||||
|
|
||||||
|
#[Object]
|
||||||
|
impl QueryRoot {
|
||||||
|
async fn value(&self) -> i32 {
|
||||||
|
999
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let schema = Schema::new(QueryRoot, EmptyMutation, EmptySubscription);
|
||||||
|
let (mut tx, rx) = mpsc::unbounded();
|
||||||
|
let mut stream = http::WebSocket::new(schema, rx, WebSocketProtocols::SubscriptionsTransportWS);
|
||||||
|
|
||||||
|
tx.send(
|
||||||
|
serde_json::to_string(&value!({
|
||||||
|
"type": "start",
|
||||||
|
"id": "1",
|
||||||
|
"payload": {
|
||||||
|
"query": "query { value }"
|
||||||
|
},
|
||||||
|
}))
|
||||||
|
.unwrap(),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
stream.next().await.unwrap().unwrap_close(),
|
||||||
|
(1011, "The handshake is not completed.".to_string())
|
||||||
|
);
|
||||||
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user