pub async fn subscribe(
        &self,
        table: &str,
        filter: &Value,
        fields: &str,
    ) -> ClientResult<Subscription> {
        let request = Self::generate_subscription(table, filter, fields);

        let mut websocket = {
            self.ensure_client().await?;
            let client_lock = self.server_info.read().await;
            let address = &client_lock.as_ref().unwrap().subscription_url;

            self.client_env
                .websocket_connect(
                    &address,
                    Some(HashMap::from_iter(
                        vec![("Sec-WebSocket-Protocol".to_owned(), "graphql-ws".to_owned())]
                            .into_iter(),
                    )),
                )
                .await?
        };

        // map stream of strings into GraphQL JSON answers
        let table = table.to_owned();
        let data_stream = websocket.receiver.filter_map(move |result| {
            let closure_table = table.clone();
            async move {
                match result {
                    Err(err) => Some(Err(err)),
                    Ok(value) => {
                        let value: Value = match serde_json::from_str(&value) {
                            Err(err) => {
                                return Some(Err(Error::invalid_server_response(format!(
                                    "Subscription answer is not a valid JSON: {}\n{}",
                                    err, value
                                ))))
                            }
                            Ok(value) => value,
                        };

                        // skip ack and keep alive messages
                        if value["type"] == "connection_ack" || value["type"] == "ka" {
                            return None;
                        }

                        // try to extract the record value from the answer
                        let record_value = &value["payload"]["data"][&closure_table];

                        if record_value.is_null() {
                            Some(Err(Error::invalid_server_response(format!(
                                "Invalid subscription answer: {}",
                                value
                            ))))
                        } else {
                            Some(Ok(record_value.clone()))
                        }
                    }
                }
            }
        });

        let id = rand::thread_rng().next_u32();

        let mut init_request = serde_json::json!({
            "type": "connection_init",
            "payload": {}
        });
        if let Some(access_key) = &self.config.access_key {
            init_request["payload"]["accessKey"] = access_key.as_str().into();
        }
        websocket.sender.send(init_request.to_string()).await?;

        let request = serde_json::json!({
            "id": id,
            "type": "start",
            "payload": {
                "query": request.query,
                "variables": request.variables,
            }
        })
        .to_string();
        websocket.sender.send(request).await?;

        let mut sender = websocket.sender;
        let unsubscribe = async move {
            let _ = sender
                .send(
                    serde_json::json!({
                        "id": id,
                        "type": "stop",
                        "payload": {}
                    })
                    .to_string(),
                )
                .await;
        };

        Ok(Subscription {
            data_stream: Box::pin(data_stream),
            unsubscribe: Box::pin(unsubscribe),
        })
    }
// Returns Stream with updates database fields by provided filter
Returns Stream with updates database fields by provided filter
Returns Stream with updates database fields by provided filter
pub async fn subscribe(
        &self,
        table: &str,
        filter: &Value,
        fields: &str,
    ) -> ClientResult<Subscription> {
        let request = Self::generate_subscription(table, filter, fields);
        let mut websocket = {
            self.ensure_client().await?;
            let client_lock = self.server_info.read().await;
            let address = &client_lock.as_ref().unwrap().subscription_url;
            self.client_env
                .websocket_connect(
                    &address,
                    Some(HashMap::from_iter(
                        vec![("Sec-WebSocket-Protocol".to_owned(), "graphql-ws".to_owned())]
                            .into_iter(),
                    )),
                )
                .await?
        };
        // map stream of strings into GraphQL JSON answers
        let table = table.to_owned();
        let data_stream = websocket.receiver.filter_map(move |result| {
            let closure_table = table.clone();
            async move {
                match result {
                    Err(err) => Some(Err(err)),
                    Ok(value) => {
                        let value: Value = match serde_json::from_str(&value) {
                            Err(err) => {
                                return Some(Err(Error::invalid_server_response(format!(
                                    "Subscription answer is not a valid JSON: {}\n{}",
                                    err, value
                                ))))
                            }
                            Ok(value) => value,
                        };
                        // skip ack and keep alive messages
                        if value["type"] == "connection_ack" || value["type"] == "ka" {
                            return None;
                        }
                        // try to extract the record value from the answer
                        let record_value = &value["payload"]["data"][&closure_table];
                        if record_value.is_null() {
                            Some(Err(Error::invalid_server_response(format!(
                                "Invalid subscription answer: {}",
                                value
                            ))))
                        } else {
                            Some(Ok(record_value.clone()))
                        }
                    }
                }
            }
        });
        let id = rand::thread_rng().next_u32();
        let mut init_request = serde_json::json!({
            "type": "connection_init",
            "payload": {}
        });
        if let Some(access_key) = &self.config.access_key {
            init_request["payload"]["accessKey"] = access_key.as_str().into();
        }
        websocket.sender.send(init_request.to_string()).await?;
        let request = serde_json::json!({
            "id": id,
            "type": "start",
            "payload": {
                "query": request.query,
                "variables": request.variables,
            }
        })
        .to_string();
        websocket.sender.send(request).await?;
        let mut sender = websocket.sender;
        let unsubscribe = async move {
            let _ = sender
                .send(
                    serde_json::json!({
                        "id": id,
                        "type": "stop",
                        "payload": {}
                    })
                    .to_string(),
                )
                .await;
        };
        Ok(Subscription {
            data_stream: Box::pin(data_stream),
            unsubscribe: Box::pin(unsubscribe),
        })
    }
pub async fn subscribe ( & self , table : & str , filter : & Value , fields : & str , ) -> ClientResult < Subscription > { let request = Self :: generate_subscription ( table , filter , fields ) ; let mut websocket = { self . ensure_client ( ) . await ? ; let client_lock = self . server_info . read ( ) . await ; let address = & client_lock . as_ref ( ) . unwrap ( ) . subscription_url ; self . client_env . websocket_connect ( & address , Some ( HashMap :: from_iter ( vec ! [ ( " " to_owned ( ) " " to_owned ( ) ) ] . into_iter ( ) , ) ) , ) . await ? } ; // map stream of strings into GraphQL JSON answers let table = table . to_owned ( ) ; let data_stream = websocket . receiver . filter_map ( move | result | { let closure_table = table . clone ( ) ; async move { match result { Err ( err ) => Some ( Err ( err ) ) , Ok ( value ) => { let value : Value = match serde_json :: from_str ( & value ) { Err ( err ) => { return Some ( Err ( Error :: invalid_server_response ( format ! ( " \n " err value ) ) ) ) } Ok ( value ) => value , } ; // skip ack and keep alive messages if value [ " " ] == " " || value [ " " ] == " " { return None ; } // try to extract the record value from the answer let record_value = & value [ " " ] [ " " ] [ & closure_table ] ; if record_value . is_null ( ) { Some ( Err ( Error :: invalid_server_response ( format ! ( " " value ) ) ) ) } else { Some ( Ok ( record_value . clone ( ) ) ) } } } } } ) ; let id = rand :: thread_rng ( ) . next_u32 ( ) ; let mut init_request = serde_json :: json ! ( { " " " " " " { } } ) ; if let Some ( access_key ) = & self . config . access_key { init_request [ " " ] [ " " ] = access_key . as_str ( ) . into ( ) ; } websocket . sender . send ( init_request . to_string ( ) ) . await ? ; let request = serde_json :: json ! ( { " " id " " " " " " { " " request query " " request variables } } ) . to_string ( ) ; websocket . sender . send ( request ) . await ? ; let mut sender = websocket . sender ; let unsubscribe = async move { let _ = sender . send ( serde_json :: json ! ( { " " id " " " " " " { } } ) . to_string ( ) , ) . await ; } ; Ok ( Subscription { data_stream : Box :: pin ( data_stream ) , unsubscribe : Box :: pin ( unsubscribe ) , } ) }
Returns Stream with updates database fields by provided filter
Returns Stream with updates database fields by provided filter
&self
table
&str
filter
&Value
fields
&str