commit a4dd7621f5b3a5a3136d7e187caa90ac38c6a402 · ptr.pet/nsid-tracker

+21 -11

server/src/jetstream.rs

···

       13
       13
        
       pub struct JetstreamClient {

     

       14
       14
        
           stream: Option<WebSocketStream<MaybeTlsStream<TcpStream>>>,

     

       15
       15
        
           tls_connector: tokio_websockets::Connector,

     

       16
       16
       -
           url: SmolStr,

     

       16
       16
       +
           urls: Vec<SmolStr>,

     

       17
       17
        
       }

     

       18
       18
        
       

     

       19
       19
        
       impl JetstreamClient {

     

       20
       20
       -
           pub fn new(url: &str) -> AppResult<Self> {

     

       20
       20
       +
           pub fn new(urls: impl IntoIterator<Item = impl Into<SmolStr>>) -> AppResult<Self> {

     

       21
       21
        
               Ok(Self {

     

       22
       22
        
                   stream: None,

     

       23
       23
        
                   tls_connector: tokio_websockets::Connector::new()?,

     

       24
       24
       -
                   url: SmolStr::new(url),

     

       24
       24
       +
                   urls: urls.into_iter().map(Into::into).collect(),

     

       25
       25
        
               })

     

       26
       26
        
           }

     

       27
       27
        
       

     

       28
       28
        
           pub async fn connect(&mut self) -> AppResult<()> {

     

       29
       29
       -
               let (stream, _) = ClientBuilder::new()

     

       30
       30
       -
                   .connector(&self.tls_connector)

     

       31
       31
       -
                   .uri(&self.url)?

     

       32
       32
       -
                   .connect()

     

       33
       33
       -
                   .await?;

     

       34
       34
       -
               self.stream = Some(stream);

     

       35
       35
       -
               tracing::info!("connected to jetstream ({})", self.url);

     

       36
       36
       -
               Ok(())

     

       29
       29
       +
               for uri in &self.urls {

     

       30
       30
       +
                   let conn_result = ClientBuilder::new()

     

       31
       31
       +
                       .connector(&self.tls_connector)

     

       32
       32
       +
                       .uri(uri)?

     

       33
       33
       +
                       .connect()

     

       34
       34
       +
                       .await;

     

       35
       35
       +
                   match conn_result {

     

       36
       36
       +
                       Ok((stream, _)) => {

     

       37
       37
       +
                           self.stream = Some(stream);

     

       38
       38
       +
                           tracing::info!("connected to jetstream {}", uri);

     

       39
       39
       +
                           return Ok(());

     

       40
       40
       +
                       }

     

       41
       41
       +
                       Err(err) => {

     

       42
       42
       +
                           tracing::error!("failed to connect to jetstream {uri}: {err}");

     

       43
       43
       +
                       }

     

       44
       44
       +
                   };

     

       45
       45
       +
               }

     

       46
       46
       +
               Err(anyhow!("failed to connect to any jetstream server").into())

     

       37
       47
        
           }

     

       38
       48
        
       

     

       39
       49
        
           // automatically retries connection, only returning error if it fails many times

+13 -8

server/src/main.rs

···

       70
       70
        
               .install_default()

     

       71
       71
        
               .expect("cant install rustls crypto provider");

     

       72
       72
        
       

     

       73
       73
       -
           let mut jetstream =

     

       74
       74
       -
               match JetstreamClient::new("wss://jetstream2.us-west.bsky.network/subscribe") {

     

       75
       75
       -
                   Ok(client) => client,

     

       76
       76
       -
                   Err(err) => {

     

       77
       77
       -
                       tracing::error!("can't create jetstream client: {err}");

     

       78
       78
       -
                       return;

     

       79
       79
       -
                   }

     

       80
       80
       -
               };

     

       73
       73
       +
           let urls = [

     

       74
       74
       +
               "wss://jetstream2.fr.hose.cam/subscribe",

     

       75
       75
       +
               "wss://jetstream.fire.hose.cam/subscribe",

     

       76
       76
       +
               "wss://jetstream1.us-west.bsky.network/subscribe",

     

       77
       77
       +
               "wss://jetstream2.us-west.bsky.network/subscribe",

     

       78
       78
       +
           ];

     

       79
       79
       +
           let mut jetstream = match JetstreamClient::new(urls) {

     

       80
       80
       +
               Ok(client) => client,

     

       81
       81
       +
               Err(err) => {

     

       82
       82
       +
                   tracing::error!("can't create jetstream client: {err}");

     

       83
       83
       +
                   return;

     

       84
       84
       +
               }

     

       85
       85
       +
           };

     

       81
       86
        
       

     

       82
       87
        
           let (event_tx, mut event_rx) = tokio::sync::mpsc::channel(1000);

     

       83
       88
        
           let consume_events = tokio::spawn({