Что такое кортеж в TStream?

TStream<T>интерфейс предназначен для моделирования потока данных, обычно показаний датчиков. T в данном случае является типом, используемым для хранения отдельного показания, но «чтение» на самом деле может означать несколько чисел (например, температура, влажность и скорость ветра), объединенных в один составной тип, который в общем упоминается здесь как «кортеж». 'ценностей.

Однако, глядя на контекст вашего примера, мы имеем дело с потоком простых показаний температуры, поэтому здесь T соответствует одному числу типа Double. Поэтому выбор «кортежа» в качестве имени переменной немного сбивает с толку (математически это 1-кортеж, но в данном случае это означает просто «число»).

В вашем коде метод filter() принимает предикат, который здесь

tuple -> !optimalTempRangeRef.get().contains(tuple)

И optimalTempRangeRef.get() возвращает Range(Double), поэтому предикат говорит "Наше значение температуры выходит за пределы оптимального диапазона?"

Из документации для Range:

contains() is used to check for containment: e.g.

 Ranges.closed(2,4).contains(2);    // returns true
 Ranges.open(2,4).contains(2);      // returns false
 Ranges.atLeast(2).contains(2);     // returns true
 Ranges.greaterThan(2).contains(2); // returns false
 Ranges.atMost(2).contains(2);      // returns true
 Ranges.lessThan(2).contains(2);    // returns false

Обновлено:

Похоже, у вашей IDE проблемы с лямбда-синтаксисом Java 8, поэтому вы можете переписать свой код, используя анонимные внутренние классы, например:

import org.apache.edgent.function.Predicate;
import org.apache.edgent.function.Consumer;


public static void main( String[] args ) throws Exception
{
    TempSensor sensor = new TempSensor();
    DirectProvider dp = new DirectProvider();
    Topology topology = dp.newTopology();

    TStream<Double> tempReadings = topology.poll( sensor, 1, TimeUnit.MILLISECONDS );
    TStream<Double> filteredStream = tempReadings.filter( new Predicate<Double>()
    {
        public boolean test( Double reading )
        {
            return !optimalTempRangeRef.get().contains( reading );
        }
    } );

    filteredStream.sink( new Consumer<Double>()
    {
        public void accept( Double reading )
        {
            System.out.println( "Temperature is out of range! "
                                + "It is " + reading + "\u00b0F!" )
        }
    } );

    tempReadings.print();

    dp.submit( topology );
}