commit f744a4807db01f4f87d8854b2b2e0d863b310f7d · nekomimi.pet/microcosm-rs

Cargo.lock

···

       2679
       2679
        
       dependencies = [

     

       2680
       2680
        
        "anyhow",

     

       2681
       2681
        
        "clap",

     

       2682
       2682
       +
        "ctrlc",

     

       2682
       2683
        
        "jetstream",

     

       2683
       2684
        
        "serde_json",

     

       2684
       2685
        
        "tokio",

     

       2686
       2686
       +
        "tokio-util",

     

       2685
       2687
        
       ]

     

       2686
       2688
        
       

     

       2687
       2689
        
       [[package]]

ufos/Cargo.toml

···

       6
       6
        
       [dependencies]

     

       7
       7
        
       anyhow = "1.0.97"

     

       8
       8
        
       clap = { version = "4.5.31", features = ["derive"] }

     

       9
       9
       +
       ctrlc = "3.4.5"

     

       9
       10
        
       jetstream = { path = "../jetstream" }

     

       10
       11
        
       serde_json = "1.0.140"

     

       11
       12
        
       tokio = { version = "1.43.0", features = ["full", "sync", "time"] }

     

       13
       13
       +
       tokio-util = "0.7.13"

+46 -12

ufos/src/main.rs

···

       2
       2
        
       use std::path::PathBuf;

     

       3
       3
        
       use std::time::{Duration, Instant};

     

       4
       4
        
       

     

       5
       5
       +
       use tokio::select;

     

       6
       6
       +
       use tokio_util::sync::CancellationToken;

     

       7
       7
       +
       

     

       5
       8
        
       use jetstream::{

     

       6
       9
        
           events::{commit::CommitEvent, JetstreamEvent::Commit},

     

       7
       10
        
           DefaultJetstreamEndpoints, JetstreamCompression, JetstreamConfig, JetstreamConnector,

     
···

       30
       33
        
               ..Default::default()

     

       31
       34
        
           };

     

       32
       35
        
       

     

       36
       36
       +
           let stay_alive = CancellationToken::new();

     

       37
       37
       +
       

     

       38
       38
       +
           ctrlc::set_handler({

     

       39
       39
       +
               let mut desperation: u8 = 0;

     

       40
       40
       +
               let stay_alive = stay_alive.clone();

     

       41
       41
       +
               move || match desperation {

     

       42
       42
       +
                   0 => {

     

       43
       43
       +
                       println!("ok, signalling shutdown...");

     

       44
       44
       +
                       stay_alive.cancel();

     

       45
       45
       +
                       desperation += 1;

     

       46
       46
       +
                   }

     

       47
       47
       +
                   1.. => panic!("fine, panicking!"),

     

       48
       48
       +
               }

     

       49
       49
       +
           })?;

     

       50
       50
       +
       

     

       33
       51
        
           let jetstream: JetstreamConnector<serde_json::Value> = JetstreamConnector::new(config)?;

     

       34
       52
        
           let receiver = jetstream.connect().await?;

     

       35
       53
        
       

     
···

       37
       55
        
       

     

       38
       56
        
           let print_throttle = Duration::from_millis(400);

     

       39
       57
        
           let mut last = Instant::now();

     

       40
       40
       -
           while let Ok(event) = receiver.recv_async().await {

     

       41
       41
       -
               if let Commit(CommitEvent::Create { commit, .. }) = event {

     

       42
       42
       -
                   let now = Instant::now();

     

       43
       43
       -
                   let since = now - last;

     

       44
       44
       -
                   if since >= print_throttle {

     

       45
       45
       -
                       let overshoot = since - print_throttle; // adjust to keep the rate on average

     

       46
       46
       -
                       last = now - overshoot;

     

       47
       47
       -
                       println!(

     

       48
       48
       -
                           "{}: {}",

     

       49
       49
       -
                           &*commit.info.collection,

     

       50
       50
       -
                           serde_json::to_string(&commit.record)?

     

       51
       51
       -
                       );

     

       58
       58
       +
           loop {

     

       59
       59
       +
               select! {

     

       60
       60
       +
                   _ = stay_alive.cancelled() => {

     

       61
       61
       +
                       eprintln!("byeeee");

     

       62
       62
       +
                       break

     

       63
       63
       +
                   }

     

       64
       64
       +
                   ev = receiver.recv_async() => {

     

       65
       65
       +
                       match ev {

     

       66
       66
       +
                           Ok(event) => {

     

       67
       67
       +
                               if let Commit(CommitEvent::Create { commit, .. }) = event {

     

       68
       68
       +
                                   let now = Instant::now();

     

       69
       69
       +
                                   let since = now - last;

     

       70
       70
       +
                                   if since >= print_throttle {

     

       71
       71
       +
                                       let overshoot = since - print_throttle; // adjust to keep the rate on average

     

       72
       72
       +
                                       last = now - overshoot;

     

       73
       73
       +
                                       println!(

     

       74
       74
       +
                                           "{}: {}",

     

       75
       75
       +
                                           &*commit.info.collection,

     

       76
       76
       +
                                           serde_json::to_string(&commit.record)?

     

       77
       77
       +
                                       );

     

       78
       78
       +
                                   }

     

       79
       79
       +
                               }

     

       80
       80
       +
                           },

     

       81
       81
       +
                           Err(e) => {

     

       82
       82
       +
                               eprintln!("jetstream event error: {e:?}");

     

       83
       83
       +
                               break

     

       84
       84
       +
                           }

     

       85
       85
       +
                       }

     

       52
       86
        
                   }

     

       53
       87
        
               }

     

       54
       88
        
           }

ufos/src/serialize.rs

This is a binary file and will not be displayed.