Fix subscriptions with custom scalars (#706)

integration_tests/juniper_tests/src/custom_scalar.rs

@@ -4,7 +4,7 @@ use juniper::{
     execute,
     parser::{ParseError, ScalarToken, Spanning, Token},
     serde::de,
-    EmptyMutation, EmptySubscription, InputValue, Object, ParseScalarResult, RootNode, ScalarValue,
+    EmptyMutation, FieldResult, InputValue, Object, ParseScalarResult, RootNode, ScalarValue,
     Value, Variables,
 };
 use std::fmt;
@@ -172,15 +172,24 @@ impl TestType {
     }
 }
 
+struct TestSubscriptionType;
+
+#[juniper::graphql_subscription(
+    Scalar = MyScalarValue
+)]
+impl TestSubscriptionType {
+    async fn foo(
+    ) -> std::pin::Pin<Box<dyn futures::Stream<Item = FieldResult<i32, MyScalarValue>> + Send>>
+    {
+        Box::pin(futures::stream::empty())
+    }
+}
+
 async fn run_variable_query<F>(query: &str, vars: Variables<MyScalarValue>, f: F)
 where
     F: Fn(&Object<MyScalarValue>) -> (),
 {
-    let schema = RootNode::new(
-        TestType,
-        EmptyMutation::<()>::new(),
-        EmptySubscription::<()>::new(),
-    );
+    let schema = RootNode::new(TestType, EmptyMutation::<()>::new(), TestSubscriptionType);
 
     let (result, errs) = execute(query, None, &schema, &vars, &())
         .await

juniper_codegen/src/util/mod.rs

@@ -1218,7 +1218,7 @@ impl GraphQLTypeDefiniton {
                             });
                             Ok(
                                 ::juniper::Value::Scalar::<
-                                    ::juniper::ValuesStream
+                                    ::juniper::ValuesStream::<#scalar>
                                 >(Box::pin(f))
                             )
                         })