···

       
94
       
 
       
		}

     

       
95
       
 
       
	}()

     

       
96
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
97
       
 
       
	return nil

     

       
98
       
 
       
}

     

       
99
       
 
       


     

···

       
94
       
 
       
		}

     

       
95
       
 
       
	}()

     

       
96
       
 
       


     

       
97
       
+
       
	log.Printf("started jetstream")

     

       
98
       
+
       


     

       
99
       
 
       
	return nil

     

       
100
       
 
       
}

     

       
101
       
 
       


     

···

       
36
       
 
       
		RawQuery: queryParams,

     

       
37
       
 
       
	}

     

       
38
       
 
       


     

       
39
       
-
       
	fmt.Println("URL:", u.String())

     

       
40
       
 
       
	return u

     

       
41
       
 
       
}

     

       
42
       
 
       


     
···

       
70
       
 
       


     

       
71
       
 
       
	var collections, dids string

     

       
72
       
 
       
	if len(j.collections) > 0 {

     

       
73
       
-
       
		collections = fmt.Sprintf("wantedCollections=%s", j.collections[0])

     

       
74
       
 
       
		for _, collection := range j.collections[1:] {

     

       
75
       
 
       
			collections += fmt.Sprintf("&wantedCollections=%s", collection)

     

       
76
       
 
       
		}

     
···

       
100
       
 
       
func (j *JetstreamClient) connect(cursor int64) error {

     

       
101
       
 
       
	queryParams := j.buildQueryParams(cursor)

     

       
102
       
 
       
	u := j.buildWebsocketURL(queryParams)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
103
       
 
       


     

       
104
       
 
       
	dialer := websocket.Dialer{

     

       
105
       
 
       
		HandshakeTimeout: 10 * time.Second,

     
···

       
130
       
 
       
			return

     

       
131
       
 
       
		case <-j.reconnectCh:

     

       
132
       
 
       
			// Reconnect with new parameters

     

       
133
       
-
       
			// cursor := time.Now().Add(-5 * time.Second).UnixMicro()

     

       
134
       
-
       
			if err := j.connect(0); err != nil {

     

       
135
       
 
       
				log.Printf("error reconnecting to jetstream: %v", err)

     

       
136
       
 
       
				return

     

       
137
       
 
       
			}

     

···

       
36
       
 
       
		RawQuery: queryParams,

     

       
37
       
 
       
	}

     

       
38
       
 
       


     

       
0
       
 
       

     

       
39
       
 
       
	return u

     

       
40
       
 
       
}

     

       
41
       
 
       


     
···

       
69
       
 
       


     

       
70
       
 
       
	var collections, dids string

     

       
71
       
 
       
	if len(j.collections) > 0 {

     

       
72
       
+
       
		collections = fmt.Sprintf("wantedCollections=%s&cursor=%d", j.collections[0], cursor)

     

       
73
       
 
       
		for _, collection := range j.collections[1:] {

     

       
74
       
 
       
			collections += fmt.Sprintf("&wantedCollections=%s", collection)

     

       
75
       
 
       
		}

     
···

       
99
       
 
       
func (j *JetstreamClient) connect(cursor int64) error {

     

       
100
       
 
       
	queryParams := j.buildQueryParams(cursor)

     

       
101
       
 
       
	u := j.buildWebsocketURL(queryParams)

     

       
102
       
+
       


     

       
103
       
+
       
	log.Printf("connecting to jetstream at: %s", u.String())

     

       
104
       
 
       


     

       
105
       
 
       
	dialer := websocket.Dialer{

     

       
106
       
 
       
		HandshakeTimeout: 10 * time.Second,

     
···

       
131
       
 
       
			return

     

       
132
       
 
       
		case <-j.reconnectCh:

     

       
133
       
 
       
			// Reconnect with new parameters

     

       
134
       
+
       
			cursor := time.Now().Add(-5 * time.Second).UnixMicro()

     

       
135
       
+
       
			if err := j.connect(cursor); err != nil {

     

       
136
       
 
       
				log.Printf("error reconnecting to jetstream: %v", err)

     

       
137
       
 
       
				return

     

       
138
       
 
       
			}