Pipes and filters

type Rec = Record<string, string | number | boolean>;
type Filter = (rows: Iterable<Rec>) => Iterable<Rec>;
const DISTRICT: Record<string, string> = { D12: "Centre", D40: "Harbour" };

// Each filter does one job, reads a stream and writes a stream. None knows about the others.
function* dropTestRides(rows: Iterable<Rec>): Iterable<Rec> { for (const r of rows) if (!r.test) yield r; }
function* capMinutes(rows: Iterable<Rec>): Iterable<Rec> { for (const r of rows) yield { ...r, minutes: Math.min(r.minutes as number, 720) }; } // forgotten bikes count as 12 hours
function* addDistrict(rows: Iterable<Rec>): Iterable<Rec> { for (const r of rows) yield { ...r, district: DISTRICT[r.dock as string] }; }
function* anonymise(rows: Iterable<Rec>): Iterable<Rec> { for (const r of rows) { const { rider: _drop, ...rest } = r; void _drop; yield rest; } }
function* countByDistrict(rows: Iterable<Rec>): Iterable<Rec> {
  const totals = new Map<string, [number, number]>();
  for (const r of rows) { const t = totals.get(r.district as string) ?? [0, 0]; t[0] += 1; t[1] += r.minutes as number; totals.set(r.district as string, t); }
  for (const d of [...totals.keys()].sort()) { const [n, m] = totals.get(d) as [number, number]; yield { district: d, rides: n, minutes: m }; }
}
function* toCsv(rows: Iterable<Rec>): Iterable<Rec> {
  yield { line: "district,rides,minutes" };
  for (const r of rows) yield { line: `${r.district},${r.rides},${r.minutes}` };
}

// The pipe: each filter's output feeds the next.
const pipeline = (source: Iterable<Rec>, ...filters: [string, Filter][]): Iterable<Rec> => filters.reduce((s, [, f]) => f(s), source);

const rides: Rec[] = [
  { id: "R1", rider: "Mia", minutes: 42, dock: "D12", test: false },
  { id: "R2", rider: "Leo", minutes: 15, dock: "D12", test: false },
  { id: "R3", rider: "qa-bot", minutes: 5, dock: "D12", test: true },
  { id: "R4", rider: "Ana", minutes: 900, dock: "D40", test: false },
];
const steps: [string, Filter][] = [["drop_test_rides", dropTestRides], ["cap_minutes", capMinutes], ["add_district", addDistrict],
  ["anonymise", anonymise], ["count_by_district", countByDistrict], ["to_csv", toCsv]];
for (const row of pipeline(rides, ...steps)) console.log(row.line);
console.log("pipeline: " + steps.map(([n]) => n).join(" | "));

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class Main {
    static final Map<String, String> DISTRICT = Map.of("D12", "Centre", "D40", "Harbour");
    record Step(String name, Function<Stream<Map<String, Object>>, Stream<Map<String, Object>>> f) {}

    static Map<String, Object> with(Map<String, Object> r, String k, Object v) { Map<String, Object> m = new HashMap<>(r); m.put(k, v); return m; }

    // Each filter does one job, reads a stream and writes a stream. None knows about the others.
    static Stream<Map<String, Object>> dropTestRides(Stream<Map<String, Object>> rows) { return rows.filter(r -> !(Boolean) r.get("test")); }
    static Stream<Map<String, Object>> capMinutes(Stream<Map<String, Object>> rows) { return rows.map(r -> with(r, "minutes", Math.min((Integer) r.get("minutes"), 720))); } // forgotten bikes count as 12 hours
    static Stream<Map<String, Object>> addDistrict(Stream<Map<String, Object>> rows) { return rows.map(r -> with(r, "district", DISTRICT.get((String) r.get("dock")))); }
    static Stream<Map<String, Object>> anonymise(Stream<Map<String, Object>> rows) { return rows.map(r -> { Map<String, Object> m = new HashMap<>(r); m.remove("rider"); return m; }); }
    static Stream<Map<String, Object>> countByDistrict(Stream<Map<String, Object>> rows) {
        Map<String, int[]> totals = new TreeMap<>();
        rows.forEach(r -> { int[] t = totals.computeIfAbsent((String) r.get("district"), k -> new int[2]); t[0]++; t[1] += (Integer) r.get("minutes"); });
        List<Map<String, Object>> out = new ArrayList<>();
        totals.forEach((d, t) -> out.add(Map.of("district", d, "rides", t[0], "minutes", t[1])));
        return out.stream();
    }
    static Stream<Map<String, Object>> toCsv(Stream<Map<String, Object>> rows) {
        return Stream.concat(Stream.of(Map.<String, Object>of("line", "district,rides,minutes")),
            rows.map(r -> Map.<String, Object>of("line", r.get("district") + "," + r.get("rides") + "," + r.get("minutes"))));
    }

    // The pipe: each filter's output feeds the next.
    static Stream<Map<String, Object>> pipeline(Stream<Map<String, Object>> source, List<Step> steps) {
        Stream<Map<String, Object>> s = source;
        for (Step st : steps) s = st.f().apply(s);
        return s;
    }

    static Map<String, Object> ride(String id, String rider, int minutes, String dock, boolean test) {
        Map<String, Object> m = new HashMap<>();
        m.put("id", id); m.put("rider", rider); m.put("minutes", minutes); m.put("dock", dock); m.put("test", test);
        return m;
    }

    public static void main(String[] args) {
        List<Map<String, Object>> rides = List.of(ride("R1", "Mia", 42, "D12", false), ride("R2", "Leo", 15, "D12", false),
            ride("R3", "qa-bot", 5, "D12", true), ride("R4", "Ana", 900, "D40", false));
        List<Step> steps = List.of(new Step("drop_test_rides", Main::dropTestRides), new Step("cap_minutes", Main::capMinutes), new Step("add_district", Main::addDistrict),
            new Step("anonymise", Main::anonymise), new Step("count_by_district", Main::countByDistrict), new Step("to_csv", Main::toCsv));
        pipeline(rides.stream(), steps).forEach(r -> System.out.println(r.get("line")));
        System.out.println("pipeline: " + steps.stream().map(Step::name).collect(Collectors.joining(" | ")));
    }
}