func (c *Consumer) twLoop(stopped <-chan none) {
	ticker := time.NewTicker(c.client.config.Metadata.RefreshFrequency / 2)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			topics, err := c.client.Topics()
			if err != nil {
				c.handleError(&Error{Ctx: "topics", error: err})
				return
			}

			for _, topic := range topics {
				if !c.isKnownCoreTopic(topic) &&
					!c.isKnownExtraTopic(topic) &&
					c.isPotentialExtraTopic(topic) {
					return
				}
			}
		case <-stopped:
			return
		case <-c.dying:
			return
		}
	}
}
func (c *Consumer) twLoop(stopped <-chan none) {
	ticker := time.NewTicker(c.client.config.Metadata.RefreshFrequency / 2)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			topics, err := c.client.Topics()
			if err != nil {
				c.handleError(&Error{Ctx: "topics", error: err})
				return
			}

			for _, topic := range topics {
				if !c.isKnownCoreTopic(topic) &&
					!c.isKnownExtraTopic(topic) &&
					c.isPotentialExtraTopic(topic) {
					return
				}
			}
		case <-stopped:
			return
		case <-c.dying:
			return
		}
	}
}
func ( c * Consumer ) twLoop ( stopped <- chan none ) { ticker := time . NewTicker ( c . client . config . Metadata . RefreshFrequency / 2 ) defer ticker . Stop ( ) for { select { case <- ticker . C : topics , err := c . client . Topics ( ) if err != nil { c . handleError ( & Error { Ctx : " " , error : err } ) return } for _ , topic := range topics { if ! c . isKnownCoreTopic ( topic ) && ! c . isKnownExtraTopic ( topic ) && c . isPotentialExtraTopic ( topic ) { return } } case <- stopped : return case <- c . dying : return } } }
// topic watcher loop, triggered by the mainLoop
topic watcher loop triggered by the mainLoop