···

       
172
       
172
       
 
       
func (c *Consumer) startConnectionLoop(ctx context.Context, source Source) {

     

       
173
       
173
       
 
       
	defer c.wg.Done()

     

       
174
       
174
       
 
       


     

       
175
       
175
       
+
       
	// attempt connection initially

     

       
176
       
176
       
+
       
	err := c.runConnection(ctx, source)

     

       
177
       
177
       
+
       
	if err != nil {

     

       
178
       
178
       
+
       
		c.logger.Error("failed to run connection", "err", err)

     

       
179
       
179
       
+
       
	}

     

       
180
       
180
       
+
       


     

       
181
       
181
       
+
       
	timer := time.NewTimer(1 * time.Minute)

     

       
182
       
182
       
+
       
	defer timer.Stop()

     

       
183
       
183
       
+
       


     

       
184
       
184
       
+
       
	// every subsequent attempt is delayed by 1 minute

     

       
175
       
185
       
 
       
	for {

     

       
176
       
186
       
 
       
		select {

     

       
177
       
187
       
 
       
		case <-ctx.Done():

     

       
178
       
188
       
 
       
			return

     

       
179
       
179
       
-
       
		default:

     

       
189
       
189
       
+
       
		case <-timer.C:

     

       
180
       
190
       
 
       
			err := c.runConnection(ctx, source)

     

       
181
       
191
       
 
       
			if err != nil {

     

       
182
       
192
       
 
       
				c.logger.Error("failed to run connection", "err", err)

     

       
183
       
193
       
 
       
			}

     

       
194
       
194
       
+
       
			timer.Reset(1 * time.Minute)

     

       
184
       
195
       
 
       
		}

     

       
185
       
196
       
 
       
	}

     

       
186
       
197
       
 
       
}