func (m *Monitor) listenForDockerEvents() {
	m.wg.Add(1)
	filter := filters.NewArgs()
	filter.Add("type", events.NetworkEventType)
	filter.Add("type", events.ContainerEventType)
	msgs, errs := m.Client.Events(m.EventsContext, types.EventsOptions{
		Filters: filter,
	})
	for {
		select {
		case err := <-errs:
			// Event stream was cancelled, shutting down
			if err == context.Canceled {
				m.wg.Done()
				return
			}
			// Re-subscribed to events stream
			dockerLogger.Errorf("Docker monitor: %v", err)
			m.eventStreamRetries++
			if m.eventStreamRetries >= m.eventStreamMaxRetries {
				dockerLogger.Errorf("Docker event monitor max retries reached")
				m.wg.Done()
				return
			}
			m.wg.Done()
			m.listenForDockerEvents()
			return
		case msg := <-msgs:
			// Reset eventStreamRetries
			m.eventStreamRetries = 0
			switch msg.Type {
			case events.ContainerEventType:
				if err := m.handleContainerEvent(msg); err != nil {
					dockerLogger.Warnf("Docker monitor error: %v", err)
				}
			case events.NetworkEventType:
				if err := m.handleNetworkEvent(msg); err != nil {
					dockerLogger.Warnf("Docker monitor error: %v", err)
				}
			default:
				dockerLogger.Warnf("unknown Docker event type: %v", msg.Type)
			}

		}
	}
}
// listenForDockerEvents - Listens for Docker events
Listens for Docker events
// listenForDockerEvents - Listens for Docker events
func (m *Monitor) listenForDockerEvents() {
	m.wg.Add(1)
	filter := filters.NewArgs()
	filter.Add("type", events.NetworkEventType)
	filter.Add("type", events.ContainerEventType)
	msgs, errs := m.Client.Events(m.EventsContext, types.EventsOptions{
		Filters: filter,
	})
	for {
		select {
		case err := <-errs:
			if err == context.Canceled {
				m.wg.Done()
				return
			}
			dockerLogger.Errorf("Docker monitor: %v", err)
			m.eventStreamRetries++
			if m.eventStreamRetries >= m.eventStreamMaxRetries {
				dockerLogger.Errorf("Docker event monitor max retries reached")
				m.wg.Done()
				return
			}
			m.wg.Done()
			m.listenForDockerEvents()
			return
		case msg := <-msgs:
			m.eventStreamRetries = 0
			switch msg.Type {
			case events.ContainerEventType:
				if err := m.handleContainerEvent(msg); err != nil {
					dockerLogger.Warnf("Docker monitor error: %v", err)
				}
			case events.NetworkEventType:
				if err := m.handleNetworkEvent(msg); err != nil {
					dockerLogger.Warnf("Docker monitor error: %v", err)
				}
			default:
				dockerLogger.Warnf("unknown Docker event type: %v", msg.Type)
			}
		}
	}
}
func ( m * Monitor ) listenForDockerEvents ( ) { m . wg . Add ( 1 ) filter := filters . NewArgs ( ) filter . Add ( " " , events . NetworkEventType ) filter . Add ( " " , events . ContainerEventType ) msgs , errs := m . Client . Events ( m . EventsContext , types . EventsOptions { Filters : filter , } ) for { select { case err := <- errs : if err == context . Canceled { m . wg . Done ( ) return } dockerLogger . Errorf ( " " , err ) m . eventStreamRetries ++ if m . eventStreamRetries >= m . eventStreamMaxRetries { dockerLogger . Errorf ( " " ) m . wg . Done ( ) return } m . wg . Done ( ) m . listenForDockerEvents ( ) return case msg := <- msgs : m . eventStreamRetries = 0 switch msg . Type { case events . ContainerEventType : if err := m . handleContainerEvent ( msg ) ; err != nil { dockerLogger . Warnf ( " " , err ) } case events . NetworkEventType : if err := m . handleNetworkEvent ( msg ) ; err != nil { dockerLogger . Warnf ( " " , err ) } default : dockerLogger . Warnf ( " " , msg . Type ) } } } }
Listens for Docker events
Listens for Docker events
// Event stream was cancelled, shutting down // Re-subscribed to events stream // Reset eventStreamRetries