// checkQueueDepth periodically checks the size of a queue to see if
// it is too large
func (s *Serf) checkQueueDepth(limit int, name string, queue *memberlist.TransmitLimitedQueue, shutdownCh chan struct{}) {
	for {
		select {
		case <-time.After(time.Second):
			numq := queue.NumQueued()
			if numq >= limit {
				s.logger.Printf("[WARN] %s queue depth: %d", name, numq)
			}
		case <-shutdownCh:
			return
		}
	}
}

func (nDB *NetworkDB) sendTableEvent(event TableEvent_Type, nid string, tname string, key string, entry *entry) error {
	tEvent := TableEvent{
		Type:      event,
		LTime:     entry.ltime,
		NodeName:  nDB.config.NodeName,
		NetworkID: nid,
		TableName: tname,
		Key:       key,
		Value:     entry.value,
	}

	raw, err := encodeMessage(MessageTypeTableEvent, &tEvent)
	if err != nil {
		return err
	}

	var broadcastQ *memberlist.TransmitLimitedQueue
	nDB.RLock()
	thisNodeNetworks, ok := nDB.networks[nDB.config.NodeName]
	if ok {
		// The network may have been removed
		network, networkOk := thisNodeNetworks[nid]
		if !networkOk {
			nDB.RUnlock()
			return nil
		}

		broadcastQ = network.tableBroadcasts
	}
	nDB.RUnlock()

	// The network may have been removed
	if broadcastQ == nil {
		return nil
	}

	broadcastQ.QueueBroadcast(&tableEventMessage{
		msg:   raw,
		id:    nid,
		tname: tname,
		key:   key,
		node:  nDB.config.NodeName,
	})
	return nil
}

// checkQueueDepth periodically checks the size of a queue to see if
// it is too large
func (s *Serf) checkQueueDepth(name string, queue *memberlist.TransmitLimitedQueue) {
	for {
		select {
		case <-time.After(time.Second):
			numq := queue.NumQueued()
			if numq >= s.config.QueueDepthWarning {
				s.logger.Printf("[WARN] %s queue depth: %d", name, numq)
			}
			if numq > s.config.MaxQueueDepth {
				s.logger.Printf("[WARN] %s queue depth (%d) exceeds limit (%d), dropping messages!",
					name, numq, s.config.MaxQueueDepth)
				queue.Prune(s.config.MaxQueueDepth)
			}
		case <-s.shutdownCh:
			return
		}
	}
}