public static List < NetFlowV9BaseRecord > parseRecords ( ByteBuf bb , Map < Integer , NetFlowV9Template > cache , NetFlowV9OptionTemplate optionTemplate ) { List < NetFlowV9BaseRecord > records = new ArrayList <> ( ) ; int flowSetId = bb . readUnsignedShort ( ) ; int length = bb . readUnsignedShort ( ) ; int end = bb . readerIndex ( ) - 4 + length ; List < NetFlowV9FieldDef > defs ; boolean isOptionTemplate = optionTemplate != null && optionTemplate . templateId ( ) == flowSetId ; if ( isOptionTemplate ) { defs = optionTemplate . optionDefs ( ) ; } else { NetFlowV9Template t = cache . get ( flowSetId ) ; if ( t == null ) { return Collections . emptyList ( ) ; } defs = t . definitions ( ) ; } // calculate record unit size int unitSize = 0 ; for ( NetFlowV9FieldDef def : defs ) { unitSize += def . length ( ) ; } while ( bb . readerIndex ( ) < end && bb . readableBytes ( ) >= unitSize ) { final ImmutableMap . Builder < String , Object > fields = ImmutableMap . builder ( ) ; for ( NetFlowV9FieldDef def : defs ) { final String key = def . type ( ) . name ( ) . toLowerCase ( Locale . ROOT ) ; final Optional < Object > optValue = def . parse ( bb ) ; optValue . ifPresent ( value -> fields . put ( key , value ) ) ; } if ( isOptionTemplate ) { final ImmutableMap . Builder < Integer , Object > scopes = ImmutableMap . builder ( ) ; for ( NetFlowV9ScopeDef def : optionTemplate . scopeDefs ( ) ) { int t = def . type ( ) ; int len = def . length ( ) ; long l = 0 ; for ( int i = 0 ; i < len ; i ++ ) { l <<= 8 ; l |= bb . readUnsignedByte ( ) ; } scopes . put ( t , l ) ; } records . add ( NetFlowV9OptionRecord . create ( fields . build ( ) , scopes . build ( ) ) ) ; } else { records . add ( NetFlowV9Record . create ( fields . build ( ) ) ) ; } // This flowset cannot contain another record, treat as padding if ( end - bb . readerIndex ( ) < unitSize ) { break ; } } bb . readerIndex ( end ) ; return records ; }
Data FlowSet Format