···

       
179
       
 
       
                }

     

       
180
       
 
       
                Err(e) => {

     

       
181
       
 
       
                    eprintln!("jetstream: could not read message from socket. closing: {e:?}");

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
182
       
 
       
                    match socket.close(None) {

     

       
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:?}"),

     

···

       
179
       
 
       
                }

     

       
180
       
 
       
                Err(e) => {

     

       
181
       
 
       
                    eprintln!("jetstream: could not read message from socket. closing: {e:?}");

     

       
182
       
+
       
                    if let TError::Io(io_err) = e {

     

       
183
       
+
       
                        if matches!(io_err.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) {

     

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

     

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

     

       
186
       
+
       
                            break;

     

       
187
       
+
       
                        }

     

       
188
       
+
       
                    }

     

       
189
       
 
       
                    match socket.close(None) {

     

       
190
       
 
       
                        Err(TError::ConnectionClosed) => {

     

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

     

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

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
193
       
 
       
                            break;

     

       
194
       
 
       
                        }

     

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