Decode decodes the flow records
(mem MemCache)
| 378 | |
| 379 | // Decode decodes the flow records |
| 380 | func (d *Decoder) Decode(mem MemCache) (*Message, error) { |
| 381 | var msg = new(Message) |
| 382 | |
| 383 | // IPFIX Message Header decoding |
| 384 | if err := msg.Header.unmarshal(d.reader); err != nil { |
| 385 | return nil, err |
| 386 | } |
| 387 | // IPFIX Message Header validation |
| 388 | if err := msg.Header.validate(); err != nil { |
| 389 | return nil, err |
| 390 | } |
| 391 | |
| 392 | // Add source IP address as Agent ID |
| 393 | msg.AgentID = d.raddr.String() |
| 394 | |
| 395 | // In case there are multiple non-fatal errors, collect them and report all of them. |
| 396 | // The rest of the received sets will still be interpreted, until a fatal error is encountered. |
| 397 | // A non-fatal error is for example an illegal data record or unknown template id. |
| 398 | var decodeErrors []error |
| 399 | for d.reader.Len() > 4 { |
| 400 | if err := d.decodeSet(mem, msg); err != nil { |
| 401 | switch err.(type) { |
| 402 | case nonfatalError: |
| 403 | decodeErrors = append(decodeErrors, err) |
| 404 | default: |
| 405 | return nil, err |
| 406 | } |
| 407 | } |
| 408 | } |
| 409 | |
| 410 | return msg, combineErrors(decodeErrors...) |
| 411 | } |
| 412 | |
| 413 | func (d *Decoder) decodeSet(mem MemCache, msg *Message) error { |
| 414 | startCount := d.reader.ReadCount() |