···

       
2
       
 
       
use metrics::{

     

       
3
       
 
       
    counter, describe_counter, describe_gauge, describe_histogram, gauge, histogram, Unit,

     

       
4
       
 
       
};

     

       
5
       
-
       
use std::io::{Cursor, Read};

     

       
6
       
 
       
use std::net::ToSocketAddrs;

     

       
7
       
 
       
use std::thread;

     

       
8
       
 
       
use std::time;

     
···

       
183
       
 
       
                        Err(TError::ConnectionClosed) => {

     

       
184
       
 
       
                            counter!("jetstream_read_fail", "url" => stream.clone(), "reason" => "clean close").increment(1);

     

       
185
       
 
       
                            println!("jetstream closed the websocket cleanly.");

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
186
       
 
       
                            break;

     

       
187
       
 
       
                        }

     

       
188
       
 
       
                        r => eprintln!("jetstream: close result after error: {r:?}"),

     

···

       
2
       
 
       
use metrics::{

     

       
3
       
 
       
    counter, describe_counter, describe_gauge, describe_histogram, gauge, histogram, Unit,

     

       
4
       
 
       
};

     

       
5
       
+
       
use std::io::{Cursor, ErrorKind, Read};

     

       
6
       
 
       
use std::net::ToSocketAddrs;

     

       
7
       
 
       
use std::thread;

     

       
8
       
 
       
use std::time;

     
···

       
183
       
 
       
                        Err(TError::ConnectionClosed) => {

     

       
184
       
 
       
                            counter!("jetstream_read_fail", "url" => stream.clone(), "reason" => "clean close").increment(1);

     

       
185
       
 
       
                            println!("jetstream closed the websocket cleanly.");

     

       
186
       
+
       
                            break;

     

       
187
       
+
       
                        }

     

       
188
       
+
       
                        Err(TError::Io(e))

     

       
189
       
+
       
                            if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) =>

     

       
190
       
+
       
                        {

     

       
191
       
+
       
                            counter!("jetstream_read_fail", "url" => stream.clone(), "reason" => "timed out").increment(1);

     

       
192
       
+
       
                            println!("jetstream socket timed out. bailing to reconnect -- should we be trying to close first?");

     

       
193
       
 
       
                            break;

     

       
194
       
 
       
                        }

     

       
195
       
 
       
                        r => eprintln!("jetstream: close result after error: {r:?}"),