synchronized void line ( String line , long arrivalTime ) { if ( count . incrementAndGet ( ) % logCountFrequency == 0 ) log . info ( "count=" + count . get ( ) + ",buffer size=" + lines . size ( ) ) ; NmeaMessage nmea ; try { nmea = NmeaUtil . parseNmea ( line ) ; } catch ( NmeaMessageParseException e ) { listener . invalidNmea ( line , arrivalTime , e . getMessage ( ) ) ; return ; } // if is multi line message then don't report to listener till last // message in sequence has been received. if ( ! nmea . isSingleSentence ( ) ) { Optional < List < NmeaMessage >> messages = nmeaBuffer . add ( nmea ) ; if ( messages . isPresent ( ) ) { Optional < NmeaMessage > joined = AisNmeaBuffer . concatenateMessages ( messages . get ( ) ) ; if ( joined . isPresent ( ) ) { if ( joined . get ( ) . getUnixTimeMillis ( ) != null ) listener . message ( joined . get ( ) . toLine ( ) , joined . get ( ) . getUnixTimeMillis ( ) ) ; else listener . message ( joined . get ( ) . toLine ( ) , arrivalTime ) ; } // TODO else report error, might need to change signature of // listener to handle problem with multi-line message } return ; } if ( nmea . getUnixTimeMillis ( ) != null ) { listener . message ( line , nmea . getUnixTimeMillis ( ) ) ; return ; } if ( ! matchWithTimestampLine ) { listener . message ( line , arrivalTime ) ; return ; } if ( ! NmeaUtil . isValid ( line ) ) return ; addLine ( line , arrivalTime ) ; log . debug ( "buffer lines=" + lines . size ( ) ) ; Integer earliestTimestampLineIndex = getEarliestTimestampLineIndex ( lines ) ; Set < Integer > removeThese ; if ( earliestTimestampLineIndex != null ) { removeThese = matchWithClosestAisMessageIfBufferLargeEnough ( arrivalTime , earliestTimestampLineIndex ) ; } else removeThese = findExpiredIndexesBeforeIndex ( lastIndex ( ) ) ; TreeSet < Integer > orderedIndexes = Sets . newTreeSet ( removeThese ) ; for ( int index : orderedIndexes . descendingSet ( ) ) { removeLineWithIndex ( index ) ; } }
Handles the arrival of a new NMEA line at the given arrival time .