pub async fn connect(&mut self) -> Result<RedisStore, ()> {
        let redis_info = self.redis_url.clone();
        let (encryption_key, decryption_key) = generate_keys(&self.secret[..]);
        self.secret.zeroize(); // clear the secret after it has been used for key generation
        let poll_interval = self.poll_interval;
        let ilp_address = self.node_ilp_address.clone();

        let client = Client::open(redis_info.clone())
            .map_err(|err| error!("Error creating subscription Redis client: {:?}", err))?;
        debug!("Connected subscription client to redis: {:?}", client);
        let mut connection = RedisReconnect::connect(redis_info.clone())
            .map_err(|_| ())
            .await?;
        let mut sub_connection = client
            .get_connection()
            .map_err(|err| error!("Error connecting subscription client to Redis: {:?}", err))?;
        // Before initializing the store, check if we have an address
        // that was configured due to adding a parent. If no parent was
        // found, use the builder's provided address (local.host) or the
        // one we decided to override it with
        let address: Option<String> = connection
            .get(&*prefixed_key(&self.db_prefix, PARENT_ILP_KEY))
            .map_err(|err| {
                error!(
                    "Error checking whether we have a parent configured: {:?}",
                    err
                )
            })
            .await?;
        let node_ilp_address = if let Some(address) = address {
            Address::from_str(&address).unwrap()
        } else {
            ilp_address
        };

        let (all_payment_publisher, _) = broadcast::channel::<PaymentNotification>(256);

        let store = RedisStore {
            ilp_address: Arc::new(RwLock::new(node_ilp_address)),
            connection,
            subscriptions: Arc::new(Mutex::new(HashMap::new())),
            payment_publisher: all_payment_publisher,
            exchange_rates: Arc::new(RwLock::new(HashMap::new())),
            routes: Arc::new(RwLock::new(Arc::new(HashMap::new()))),
            encryption_key: Arc::new(encryption_key),
            decryption_key: Arc::new(decryption_key),
            db_prefix: self.db_prefix.clone(),
        };

        // Poll for routing table updates
        // Note: if this behavior changes, make sure to update the Drop implementation
        let connection_clone = Arc::downgrade(&store.connection.conn);
        let redis_info = store.connection.redis_info.clone();
        let routing_table = store.routes.clone();

        let db_prefix = self.db_prefix.clone();
        let poll_routes = async move {
            let mut interval = tokio::time::interval(Duration::from_millis(poll_interval));
            // Irrefutable while pattern, can we do something here?
            loop {
                interval.tick().await;
                if let Some(conn) = connection_clone.upgrade() {
                    let _ = update_routes(
                        RedisReconnect {
                            conn,
                            redis_info: redis_info.clone(),
                        },
                        routing_table.clone(),
                        &db_prefix,
                    )
                    .map_err(|err| error!("{}", err))
                    .await;
                } else {
                    debug!("Not polling routes anymore because connection was closed");
                    break;
                }
            }
            Ok::<(), ()>(())
        };
        tokio::spawn(poll_routes);

        // Here we spawn a worker thread to listen for incoming messages on Redis pub/sub,
        // running a callback for each message received.
        // This currently must be a thread rather than a task due to the redis-rs driver
        // not yet supporting asynchronous subscriptions (see https://github.com/mitsuhiko/redis-rs/issues/183).
        let subscriptions_clone = store.subscriptions.clone();
        let payment_publisher = store.payment_publisher.clone();
        let db_prefix = prefixed_key(&self.db_prefix, STREAM_NOTIFICATIONS_PREFIX).into_owned();

        // add a oneshot to provide some synchronization on a busy continious integration server
        // between this "thread of execution" and the launched listener.
        let (tx, rx) = tokio::sync::oneshot::channel();

        std::thread::spawn(move || {
            // our notifications will be PUBLISH'd to topics under this prefix
            let prefix = format!("{}*", &db_prefix);
            tx.send(()).expect("exiting as parent has exited");
            let sub_status =
                sub_connection.psubscribe::<_, _, Vec<String>>(&[prefix], move |msg| {
                    let channel_name = msg.get_channel_name();
                    if let Some(suffix) = channel_name.strip_prefix(&db_prefix) {
                        if let Ok(account_id) = Uuid::from_str(suffix) {
                            let message: PaymentNotification = match serde_json::from_slice(msg.get_payload_bytes()) {
                                Ok(s) => s,
                                Err(e) => {
                                    error!("Failed to get payload from subscription: {}", e);
                                    return ControlFlow::Continue;
                                }
                            };
                            trace!("Subscribed message received for account {}: {:?}", account_id, message);
                            if payment_publisher.receiver_count() > 0 {
                                if let Err(err) = payment_publisher.send(message.clone()) {
                                    error!("Failed to send a node-wide payment notification: {:?}", err);
                                }
                            }
                            match subscriptions_clone.lock().get_mut(&account_id) {
                                  Some(senders) => {
                                    senders.retain(|sender| {
                                        if let Err(err) = sender.unbounded_send(message.clone()) {
                                            debug!("Failed to send message: {}", err);
                                            false
                                        } else {
                                            true
                                        }
                                    });
                                },
                                None => trace!("Ignoring message for account {} because there were no open subscriptions", account_id),
                            }
                        } else {
                            error!("Invalid Uuid in channel name: {}", channel_name);
                        }
                    } else {
                        warn!("Ignoring unexpected message from Redis subscription for channel: {}", channel_name);
                    }
                    ControlFlow::Continue
                });
            match sub_status {
                Err(e) => warn!("Could not issue psubscribe to Redis: {}", e),
                Ok(_) => debug!("Successfully subscribed to Redis pubsub"),
            }
        });

        // this will at least wait until the thread has started and done a thing.
        rx.await.expect("psubscribe thread failed to launch");

        Ok(store)
    }
/// Connects to the Redis Store
///
/// Specifically
/// 1. Generates encryption and decryption keys
/// 1. Connects to the redis store (ensuring that it reconnects in case of drop)
/// 1. Gets the Node address assigned to us by our parent (if it exists)
/// 1. Starts polling for routing table updates
/// 1. Spawns a thread to notify incoming payments over WebSockets
Connects to the Redis Store
Specifically
1. Generates encryption and decryption keys
1. Connects to the redis store (ensuring that it reconnects in case of drop)
1. Gets the Node address assigned to us by our parent (if it exists)
1. Starts polling for routing table updates
1. Spawns a thread to notify incoming payments over WebSockets
Connects to the Redis Store Specifically 1 . Generates encryption and decryption keys 1 . Connects to the redis store ( ensuring that it reconnects in case of drop ) 1 . Gets the Node address assigned to us by our parent ( if it exists ) 1 . Starts polling for routing table updates 1 . Spawns a thread to notify incoming payments over WebSockets
pub async fn connect(&mut self) -> Result<RedisStore, ()> {
        let redis_info = self.redis_url.clone();
        let (encryption_key, decryption_key) = generate_keys(&self.secret[..]);
        self.secret.zeroize(); // clear the secret after it has been used for key generation
        let poll_interval = self.poll_interval;
        let ilp_address = self.node_ilp_address.clone();
        let client = Client::open(redis_info.clone())
            .map_err(|err| error!("Error creating subscription Redis client: {:?}", err))?;
        debug!("Connected subscription client to redis: {:?}", client);
        let mut connection = RedisReconnect::connect(redis_info.clone())
            .map_err(|_| ())
            .await?;
        let mut sub_connection = client
            .get_connection()
            .map_err(|err| error!("Error connecting subscription client to Redis: {:?}", err))?;
        // Before initializing the store, check if we have an address
        // that was configured due to adding a parent. If no parent was
        // found, use the builder's provided address (local.host) or the
        // one we decided to override it with
        let address: Option<String> = connection
            .get(&*prefixed_key(&self.db_prefix, PARENT_ILP_KEY))
            .map_err(|err| {
                error!(
                    "Error checking whether we have a parent configured: {:?}",
                    err
                )
            })
            .await?;
        let node_ilp_address = if let Some(address) = address {
            Address::from_str(&address).unwrap()
        } else {
            ilp_address
        };
        let (all_payment_publisher, _) = broadcast::channel::<PaymentNotification>(256);
        let store = RedisStore {
            ilp_address: Arc::new(RwLock::new(node_ilp_address)),
            connection,
            subscriptions: Arc::new(Mutex::new(HashMap::new())),
            payment_publisher: all_payment_publisher,
            exchange_rates: Arc::new(RwLock::new(HashMap::new())),
            routes: Arc::new(RwLock::new(Arc::new(HashMap::new()))),
            encryption_key: Arc::new(encryption_key),
            decryption_key: Arc::new(decryption_key),
            db_prefix: self.db_prefix.clone(),
        };
        // Poll for routing table updates
        // Note: if this behavior changes, make sure to update the Drop implementation
        let connection_clone = Arc::downgrade(&store.connection.conn);
        let redis_info = store.connection.redis_info.clone();
        let routing_table = store.routes.clone();
        let db_prefix = self.db_prefix.clone();
        let poll_routes = async move {
            let mut interval = tokio::time::interval(Duration::from_millis(poll_interval));
            // Irrefutable while pattern, can we do something here?
            loop {
                interval.tick().await;
                if let Some(conn) = connection_clone.upgrade() {
                    let _ = update_routes(
                        RedisReconnect {
                            conn,
                            redis_info: redis_info.clone(),
                        },
                        routing_table.clone(),
                        &db_prefix,
                    )
                    .map_err(|err| error!("{}", err))
                    .await;
                } else {
                    debug!("Not polling routes anymore because connection was closed");
                    break;
                }
            }
            Ok::<(), ()>(())
        };
        tokio::spawn(poll_routes);
        // Here we spawn a worker thread to listen for incoming messages on Redis pub/sub,
        // running a callback for each message received.
        // This currently must be a thread rather than a task due to the redis-rs driver
        // not yet supporting asynchronous subscriptions (see https://github.com/mitsuhiko/redis-rs/issues/183).
        let subscriptions_clone = store.subscriptions.clone();
        let payment_publisher = store.payment_publisher.clone();
        let db_prefix = prefixed_key(&self.db_prefix, STREAM_NOTIFICATIONS_PREFIX).into_owned();
        // add a oneshot to provide some synchronization on a busy continious integration server
        // between this "thread of execution" and the launched listener.
        let (tx, rx) = tokio::sync::oneshot::channel();
        std::thread::spawn(move || {
            // our notifications will be PUBLISH'd to topics under this prefix
            let prefix = format!("{}*", &db_prefix);
            tx.send(()).expect("exiting as parent has exited");
            let sub_status =
                sub_connection.psubscribe::<_, _, Vec<String>>(&[prefix], move |msg| {
                    let channel_name = msg.get_channel_name();
                    if let Some(suffix) = channel_name.strip_prefix(&db_prefix) {
                        if let Ok(account_id) = Uuid::from_str(suffix) {
                            let message: PaymentNotification = match serde_json::from_slice(msg.get_payload_bytes()) {
                                Ok(s) => s,
                                Err(e) => {
                                    error!("Failed to get payload from subscription: {}", e);
                                    return ControlFlow::Continue;
                                }
                            };
                            trace!("Subscribed message received for account {}: {:?}", account_id, message);
                            if payment_publisher.receiver_count() > 0 {
                                if let Err(err) = payment_publisher.send(message.clone()) {
                                    error!("Failed to send a node-wide payment notification: {:?}", err);
                                }
                            }
                            match subscriptions_clone.lock().get_mut(&account_id) {
                                  Some(senders) => {
                                    senders.retain(|sender| {
                                        if let Err(err) = sender.unbounded_send(message.clone()) {
                                            debug!("Failed to send message: {}", err);
                                            false
                                        } else {
                                            true
                                        }
                                    });
                                },
                                None => trace!("Ignoring message for account {} because there were no open subscriptions", account_id),
                            }
                        } else {
                            error!("Invalid Uuid in channel name: {}", channel_name);
                        }
                    } else {
                        warn!("Ignoring unexpected message from Redis subscription for channel: {}", channel_name);
                    }
                    ControlFlow::Continue
                });
            match sub_status {
                Err(e) => warn!("Could not issue psubscribe to Redis: {}", e),
                Ok(_) => debug!("Successfully subscribed to Redis pubsub"),
            }
        });
        // this will at least wait until the thread has started and done a thing.
        rx.await.expect("psubscribe thread failed to launch");
        Ok(store)
    }
pub async fn connect ( & mut self ) -> Result < RedisStore , ( ) > { let redis_info = self . redis_url . clone ( ) ; let ( encryption_key , decryption_key ) = generate_keys ( & self . secret [ .. ] ) ; self . secret . zeroize ( ) ; // clear the secret after it has been used for key generation let poll_interval = self . poll_interval ; let ilp_address = self . node_ilp_address . clone ( ) ; let client = Client :: open ( redis_info . clone ( ) ) . map_err ( | err | error ! ( " " err ) ) ? ; debug ! ( " " client ) ; let mut connection = RedisReconnect :: connect ( redis_info . clone ( ) ) . map_err ( | _ | ( ) ) . await ? ; let mut sub_connection = client . get_connection ( ) . map_err ( | err | error ! ( " " err ) ) ? ; // Before initializing the store, check if we have an address // that was configured due to adding a parent. If no parent was // found, use the builder's provided address (local.host) or the // one we decided to override it with let address : Option < String > = connection . get ( & * prefixed_key ( & self . db_prefix , PARENT_ILP_KEY ) ) . map_err ( | err | { error ! ( " " err ) } ) . await ? ; let node_ilp_address = if let Some ( address ) = address { Address :: from_str ( & address ) . unwrap ( ) } else { ilp_address } ; let ( all_payment_publisher , _ ) = broadcast :: channel :: < PaymentNotification > ( 256 ) ; let store = RedisStore { ilp_address : Arc :: new ( RwLock :: new ( node_ilp_address ) ) , connection , subscriptions : Arc :: new ( Mutex :: new ( HashMap :: new ( ) ) ) , payment_publisher : all_payment_publisher , exchange_rates : Arc :: new ( RwLock :: new ( HashMap :: new ( ) ) ) , routes : Arc :: new ( RwLock :: new ( Arc :: new ( HashMap :: new ( ) ) ) ) , encryption_key : Arc :: new ( encryption_key ) , decryption_key : Arc :: new ( decryption_key ) , db_prefix : self . db_prefix . clone ( ) , } ; // Poll for routing table updates // Note: if this behavior changes, make sure to update the Drop implementation let connection_clone = Arc :: downgrade ( & store . connection . conn ) ; let redis_info = store . connection . redis_info . clone ( ) ; let routing_table = store . routes . clone ( ) ; let db_prefix = self . db_prefix . clone ( ) ; let poll_routes = async move { let mut interval = tokio :: time :: interval ( Duration :: from_millis ( poll_interval ) ) ; // Irrefutable while pattern, can we do something here? loop { interval . tick ( ) . await ; if let Some ( conn ) = connection_clone . upgrade ( ) { let _ = update_routes ( RedisReconnect { conn , redis_info : redis_info . clone ( ) , } , routing_table . clone ( ) , & db_prefix , ) . map_err ( | err | error ! ( " " err ) ) . await ; } else { debug ! ( " " ) ; break ; } } Ok :: < ( ) , ( ) > ( ( ) ) } ; tokio :: spawn ( poll_routes ) ; // Here we spawn a worker thread to listen for incoming messages on Redis pub/sub, // running a callback for each message received. // This currently must be a thread rather than a task due to the redis-rs driver // not yet supporting asynchronous subscriptions (see https://github.com/mitsuhiko/redis-rs/issues/183). let subscriptions_clone = store . subscriptions . clone ( ) ; let payment_publisher = store . payment_publisher . clone ( ) ; let db_prefix = prefixed_key ( & self . db_prefix , STREAM_NOTIFICATIONS_PREFIX ) . into_owned ( ) ; // add a oneshot to provide some synchronization on a busy continious integration server // between this "thread of execution" and the launched listener. let ( tx , rx ) = tokio :: sync :: oneshot :: channel ( ) ; std :: thread :: spawn ( move | | { // our notifications will be PUBLISH'd to topics under this prefix let prefix = format ! ( " " db_prefix ) ; tx . send ( ( ) ) . expect ( " " ) ; let sub_status = sub_connection . psubscribe :: < _ , _ , Vec < String > > ( & [ prefix ] , move | msg | { let channel_name = msg . get_channel_name ( ) ; if let Some ( suffix ) = channel_name . strip_prefix ( & db_prefix ) { if let Ok ( account_id ) = Uuid :: from_str ( suffix ) { let message : PaymentNotification = match serde_json :: from_slice ( msg . get_payload_bytes ( ) ) { Ok ( s ) => s , Err ( e ) => { error ! ( " " e ) ; return ControlFlow :: Continue ; } } ; trace ! ( " " account_id message ) ; if payment_publisher . receiver_count ( ) > 0 { if let Err ( err ) = payment_publisher . send ( message . clone ( ) ) { error ! ( " " err ) ; } } match subscriptions_clone . lock ( ) . get_mut ( & account_id ) { Some ( senders ) => { senders . retain ( | sender | { if let Err ( err ) = sender . unbounded_send ( message . clone ( ) ) { debug ! ( " " err ) ; false } else { true } } ) ; } , None => trace ! ( " " account_id ) , } } else { error ! ( " " channel_name ) ; } } else { warn ! ( " " channel_name ) ; } ControlFlow :: Continue } ) ; match sub_status { Err ( e ) => warn ! ( " " e ) , Ok ( _ ) => debug ! ( " " ) , } } ) ; // this will at least wait until the thread has started and done a thing. rx . await . expect ( " " ) ; Ok ( store ) }
Connects to the Redis Store
Specifically
1.
Connects to the Redis Store Specifically 1 .
&mut self
&mut self