WIP - compatibility with GAT branch of SDK

Author: sd2k

backend/src/data.rs

@@ -1,9 +1,6 @@
 use std::{collections::HashMap, sync::Arc};
 
-use futures_util::{
-    stream::{FuturesOrdered, FuturesUnordered},
-    StreamExt,
-};
+use futures_util::stream::FuturesOrdered;
 
 use grafana_plugin_sdk::backend;
 use tokio::sync::RwLock;
@@ -35,14 +32,14 @@ impl backend::DataQueryError for QueryError {
 // GATs, since the `DataService::Stream` associated type can't contain references.
 // Ideally we'd just borrow the query/uid etc but it's really not a big deal.
 async fn query_data_single(
-    client: Client,
-    uid: String,
+    client: &Client,
+    uid: &str,
     query: backend::DataQuery<Query>,
     queries: Arc<RwLock<HashMap<path::QueryId, SelectStatement>>>,
 ) -> Result<backend::DataResponse, Error> {
     let q = query.query;
     let target = q.as_tail()?;
-    let rows = target.select_all(&client).await?;
+    let rows = target.select_all(client).await?;
     let mut frame = rows_to_frame(&rows);
 
     if let TailTarget::Select { statement } = target {
@@ -66,38 +63,42 @@ async fn query_data_single(
 impl backend::DataService for MaterializePlugin {
     type Query = Query;
     type QueryError = QueryError;
-    type Stream = backend::BoxDataResponseStream<Self::QueryError>;
+    type Stream<'a> = backend::BoxDataResponseStream<'a, Self::QueryError>;
 
-    async fn query_data(&self, request: backend::QueryDataRequest<Self::Query>) -> Self::Stream {
+    async fn query_data(
+        &self,
+        request: backend::QueryDataRequest<Self::Query>,
+    ) -> Self::Stream<'_> {
         let datasource_settings = request
             .plugin_context
             .datasource_instance_settings
             .clone()
             .ok_or(Error::MissingDatasource)
             .unwrap();
-        let clients: Vec<_> = request
-            .queries
-            .iter()
-            .map(|_| self.get_client(&datasource_settings))
-            .collect::<FuturesUnordered<_>>()
-            .collect()
-            .await;
+        // let clients: Vec<_> = request
+        //     .queries
+        //     .iter()
+        //     .map(|_| self.get_client(&datasource_settings))
+        //     .collect::<FuturesUnordered<_>>()
+        //     .collect()
+        //     .await;
+        let client = Arc::new(self.get_client(&datasource_settings).await.unwrap());
         let queries = self.sql_queries.clone();
         Box::pin(
             request
                 .queries
                 .into_iter()
-                .zip(clients)
-                .map(move |(x, client)| {
+                .map(|x| {
+                    let client = client.clone();
                     let queries = queries.clone();
                     let ref_id = x.ref_id.clone();
                     let uid = datasource_settings.uid.clone();
-                    async {
-                        let client = client.map_err(|source| QueryError {
-                            ref_id: ref_id.clone(),
-                            source,
-                        })?;
-                        query_data_single(client, uid, x, queries)
+                    async move {
+                        // let client = client.map_err(|source| QueryError {
+                        //     ref_id: ref_id.clone(),
+                        //     source,
+                        // })?;
+                        query_data_single(&client, &uid, x, queries)
                             .await
                             .map_err(|source| QueryError { ref_id, source })
                     }