···

       
21
       
21
       
 
       
    async with websockets.connect(relay_url, ping_timeout=None) as firehose:

     

       
22
       
22
       
 
       
        current_minute = None

     

       
23
       
23
       
 
       
        while True:

     

       
24
       
24
       
-
       
            message = BytesIO(await firehose.recv())

     

       
25
       
25
       
-
       
            header = dag_cbor.decode(message, allow_concat=True)

     

       
24
       
24
       
+
       
            frame = BytesIO(await firehose.recv())

     

       
25
       
25
       
+
       
            header = dag_cbor.decode(frame, allow_concat=True)

     

       
26
       
26
       
 
       
            if header['op'] != 1:

     

       
27
       
27
       
 
       
                continue

     

       
28
       
28
       
 
       


     

       
29
       
29
       
-
       
            redis_cnx.publish('bsky-tools:firehose:stream', message.getvalue())

     

       
29
       
29
       
+
       
            redis_cnx.publish('bsky-tools:firehose:stream', frame.getvalue())

     

       
30
       
30
       
 
       


     

       
31
       
31
       
 
       
            # checkpoint the seq

     

       
32
       
32
       
 
       
            now = datetime.now(timezone.utc)

     

       
33
       
33
       
 
       
            if now.time().minute != current_minute:

     

       
34
       
34
       
 
       
                current_minute = now.time().minute

     

       
35
       
35
       
 
       


     

       
36
       
36
       
-
       
                payload = dag_cbor.decode(message)

     

       
36
       
36
       
+
       
                payload = dag_cbor.decode(frame)

     

       
37
       
37
       
 
       
                payload_seq = payload['seq']

     

       
38
       
38
       
 
       
                payload_time = datetime.strptime(payload['time'], '%Y-%m-%dT%H:%M:%S.%fZ').replace(tzinfo=timezone.utc)

     

       
39
       
39
       
 
       
                payload_lag = now - payload_time