nl_tokens: Process { @link MetricTimeSeries } and { @link MetricValue } filtering out data point earlier or equal to timestamp that was marked as already sent .
pl_tokens: public List < SignalFxProtocolBuffers . DataPoint > process ( MetricTimeSeries mts , List < MetricValue > metricValues ) { Long lastTimestamp = mtsToLastTimestamp . get ( mts ) ; Long latestTimestamp = lastTimestamp ; List < SignalFxProtocolBuffers . DataPoint > dataPoints = new LinkedList <> ( ) ; SignalFxProtocolBuffers . DataPoint . Builder dataPointBuilder = SignalFxProtocolBuffers . DataPoint . newBuilder ( ) . setMetric ( mts . metricName ) ; if ( mts . dimensions != null ) { for ( Map . Entry < String , String > entry : mts . dimensions . entrySet ( ) ) { dataPointBuilder . addDimensions ( SignalFxProtocolBuffers . Dimension . newBuilder ( ) . setKey ( entry . getKey ( ) ) . setValue ( entry . getValue ( ) ) ) ; } } for ( MetricValue metricValue : metricValues ) { if ( lastTimestamp == null || metricValue . startTimeInMillis > lastTimestamp ) { SignalFxProtocolBuffers . DataPoint dataPoint = dataPointBuilder . setTimestamp ( metricValue . startTimeInMillis ) . setValue ( SignalFxProtocolBuffers . Datum . newBuilder ( ) . setIntValue ( metricValue . value ) ) . build ( ) ; dataPoints . add ( dataPoint ) ; latestTimestamp = lastTimestamp == null ? metricValue . startTimeInMillis : Math . <mask> ( metricValue . startTimeInMillis , latestTimestamp ) ; } } if ( lastTimestamp == null || latestTimestamp > lastTimestamp ) { mtsToLastTimestamp . put ( mts , latestTimestamp ) ; } return dataPoints ; }