···

       
425
       
 
       
            Some(Ok(message)) => {

     

       
426
       
 
       
                match message {

     

       
427
       
 
       
                    Message::Text(json) => {

     

       
428
       
-
       
                        let event: JetstreamEvent = serde_json::from_str(&json)

     

       
429
       
-
       
                            .map_err(JetstreamEventError::ReceivedMalformedJSON)?;

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
430
       
 
       
                        let event_cursor = event.cursor;

     

       
431
       
 
       


     

       
432
       
 
       
                        if let Some(last) = last_cursor {

     
···

       
456
       
 
       
                        )

     

       
457
       
 
       
                        .map_err(JetstreamEventError::CompressionDictionaryError)?;

     

       
458
       
 
       


     

       
459
       
-
       
                        let event: JetstreamEvent = serde_json::from_reader(decoder)

     

       
460
       
-
       
                            .map_err(JetstreamEventError::ReceivedMalformedJSON)?;

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
461
       
 
       
                        let event_cursor = event.cursor;

     

       
462
       
 
       


     

       
463
       
 
       
                        if let Some(last) = last_cursor {

     

···

       
425
       
 
       
            Some(Ok(message)) => {

     

       
426
       
 
       
                match message {

     

       
427
       
 
       
                    Message::Text(json) => {

     

       
428
       
+
       
                        let event: JetstreamEvent = match serde_json::from_str(&json) {

     

       
429
       
+
       
                            Ok(ev) => ev,

     

       
430
       
+
       
                            Err(e) => {

     

       
431
       
+
       
                                log::warn!(

     

       
432
       
+
       
                                    "failed to parse json: {e:?} (from {})",

     

       
433
       
+
       
                                    json.get(..24).unwrap_or(&json)

     

       
434
       
+
       
                                );

     

       
435
       
+
       
                                continue;

     

       
436
       
+
       
                            }

     

       
437
       
+
       
                        };

     

       
438
       
 
       
                        let event_cursor = event.cursor;

     

       
439
       
 
       


     

       
440
       
 
       
                        if let Some(last) = last_cursor {

     
···

       
464
       
 
       
                        )

     

       
465
       
 
       
                        .map_err(JetstreamEventError::CompressionDictionaryError)?;

     

       
466
       
 
       


     

       
467
       
+
       
                        let event: JetstreamEvent = match serde_json::from_reader(decoder) {

     

       
468
       
+
       
                            Ok(ev) => ev,

     

       
469
       
+
       
                            Err(e) => {

     

       
470
       
+
       
                                log::warn!("failed to parse json: {e:?}");

     

       
471
       
+
       
                                continue;

     

       
472
       
+
       
                            }

     

       
473
       
+
       
                        };

     

       
474
       
 
       
                        let event_cursor = event.cursor;

     

       
475
       
 
       


     

       
476
       
 
       
                        if let Some(last) = last_cursor {