Code snippet from The Coding Café - snippet-10.java

snippet-10.java

record PeakValley(String type, double value, int index) {}

public static Gatherer<Double, ?, PeakValley> peakValleyDetection() {
    return Gatherer.of(
        () -> new Object() {
            Double prev = null;      // 1️⃣ Previous value
            Double current = null;   // 2️⃣ Current value
            int index = 0;          // 3️⃣ Track position
        },
        (state, next, downstream) -> {
            if (state.prev != null && state.current != null) {  // 4️⃣ Have 3 points?
                // Check for peak: current > both neighbors
                if (state.current > state.prev && state.current > next) {
                    downstream.push(new PeakValley("PEAK", state.current, state.index));
                } 
                // Check for valley: current < both neighbors
                else if (state.current < state.prev && state.current < next) {
                    downstream.push(new PeakValley("VALLEY", state.current, state.index));
                }
            }
            // 5️⃣ Slide the window forward
            state.prev = state.current;
            state.current = next;
            state.index++;
            return true;
        }
    );
}

// Visual example:
//     Peak
//      ↓
//    103  
//   /   \
// 101   99  ← Valley
//         \
//         102