Method sFlowWorker

vflow/sflow.go:175–240

(wQuit chan struct{})

}

func (s *SFlow) sFlowWorker(wQuit chan struct{}) {
	var (
		reader *bytes.Reader
		msg    SFUDPMsg
		mirror SFUDPMsg
		ok     bool
		b      []byte
	)

LOOP:
	for {

		select {
		case <-wQuit:
			break LOOP
		case msg, ok = <-sFlowUDPCh:
			if !ok {
				break LOOP
			}
		}

		if opts.Verbose {
			logger.Printf("rcvd sflow data from: %s, size: %d bytes",
				msg.raddr, len(msg.body))
		}

		if sFlowMirrorEnabled {
			mirror.raddr = msg.raddr
			mirror.body = sFlowBuffer.Get().([]byte)
			mirror.body = append(mirror.body[:0], msg.body...)

			select {
			case sFlowMCh <- mirror:
			default:
			}
		}

		reader = bytes.NewReader(msg.body)
		d := sflow.NewSFDecoder(reader, opts.SFlowTypeFilter)
		datagram, err := d.SFDecode()
		if err != nil || (len(datagram.Counters) < 1 && len(datagram.Samples) < 1) {
			sFlowBuffer.Put(msg.body[:opts.SFlowUDPSize])
			continue
		}

		b, err = json.Marshal(datagram)
		if err != nil {
			sFlowBuffer.Put(msg.body[:opts.SFlowUDPSize])
			logger.Println(err)
			continue
		}

		atomic.AddUint64(&s.stats.DecodedCount, 1)

		if opts.Verbose {
			logger.Println(string(b))
		}