Decode decodes the flow records
()
| 282 | |
| 283 | // Decode decodes the flow records |
| 284 | func (d *Decoder) Decode() (*Message, error) { |
| 285 | var msg = new(Message) |
| 286 | |
| 287 | // Decode the Packet Header |
| 288 | if err := msg.Header.unmarshal(d.reader); err != nil { |
| 289 | return nil, err |
| 290 | } |
| 291 | // Validate the Packet Header |
| 292 | if err := msg.Header.validate(); err != nil { |
| 293 | return nil, err |
| 294 | } |
| 295 | |
| 296 | // Add source IP address as Agent ID |
| 297 | msg.AgentID = d.raddr.String() |
| 298 | |
| 299 | // Decode the Flows |
| 300 | var decodeErrors []error |
| 301 | flowCount := int(msg.Header.Count) |
| 302 | if err := d.decodeFlows(flowCount, msg); err != nil { |
| 303 | switch err.(type) { |
| 304 | case nonfatalError: |
| 305 | decodeErrors = append(decodeErrors, err) |
| 306 | default: |
| 307 | return nil, err |
| 308 | } |
| 309 | } |
| 310 | |
| 311 | return msg, combineErrors(decodeErrors...) |
| 312 | |
| 313 | } |
| 314 | |
| 315 | func (d *Decoder) decodeFlows(flowCount int, msg *Message) error { |
| 316 | remainingLen := d.reader.Len() |