Method run

vflow/ipfix.go:91–169

()

}

func (i *IPFIX) run() {
	// exit if the ipfix is disabled
	if !opts.IPFIXEnabled {
		logger.Println("ipfix has been disabled")
		return
	}

	i.pool = make(chan chan struct{}, maxWorkers)

	hostPort := net.JoinHostPort(i.addr, strconv.Itoa(i.port))
	udpAddr, _ := net.ResolveUDPAddr("udp", hostPort)

	conn, err := net.ListenUDP("udp", udpAddr)
	if err != nil {
		logger.Fatal(err)
	}

	atomic.AddInt32(&i.stats.Workers, int32(i.workers))
	for n := 0; n < i.workers; n++ {
		go func() {
			wQuit := make(chan struct{})
			i.pool <- wQuit
			i.ipfixWorker(wQuit)
		}()
	}

	logger.Printf("ipfix is running (UDP: listening on [::]:%d workers#: %d)", i.port, i.workers)

	err = ipfix.LoadExtElements(opts.VFlowConfigPath)
	if err != nil {
		logger.Println("load.ext.elements:", err)
	}

	mCache = ipfix.GetCache(opts.IPFIXTplCacheFile)
	go ipfix.RPC(mCache, &ipfix.RPCConfig{
		Enabled: opts.IPFIXRPCEnabled,
		Logger:  logger,
	})

	go mirrorIPFIXDispatcher(ipfixMCh)

	go func() {
		if !opts.ProducerEnabled {
			return
		}

		p := producer.NewProducer(opts.MQName)
		p.MQConfigFile = path.Join(opts.VFlowConfigPath, opts.MQConfigFile)
		p.MQErrorCount = &i.stats.MQErrorCount
		p.Logger = logger
		p.Chan = ipfixMQCh
		p.Topic = opts.IPFIXTopic

		if err := p.Run(); err != nil {
			logger.Fatal(err)
		}
	}()