Pipes and Filters

In the previous three variants, data crossed the boundary in one leap: the model result came back in a single call, formatting happened in one place. The pricing context’s real workflow differs. The daily shipment list passes each record through five steps: normalizing field names, deriving the zone, applying the tariff, subtracting the discount, rounding the amount. Today these steps sit inside a single body; adding a step — a fuel surcharge — means editing that body, and none of the steps can run on its own.

The pipeline style breaks this work into steps. Each step is a filter: it takes a record as input, gives a record as output, and does not know its neighbors. The connection between steps is the pipe; the chain’s order lives not in the filters but wherever the chain is built. The Pipeline lesson of the Shell Programming course built an instance: there the filter was a process, the pipe a memory buffer; here the filter is a module, the pipe a function call. Shell syntax belongs there.

The quality attribute here is maintainability, and the scenario question is: when a new step is added, how many files are edited and how many field names does that file recognize. Two more measures come alongside: fields crossing filter boundaries, and live records, in the performance efficiency family.

Five Steps in a Single Body

The first arrangement is today’s situation: a single function applies the five steps to each row in sequence. The transition between steps is not a boundary but the next statement in the same body.

mkdir -p single pipe

// single/body.mjs — five steps inside a single body: field names, zone, tariff, discount, rounding
const ZONE = { "34": "near", "06": "mid", "35": "mid", "65": "far" };
const COEFFICIENT = { near: 1, mid: 1.4, far: 2.1 };
const MINIMUM = { near: 30, mid: 45, far: 70 };
const CONTRACT = { "MUS-1": 0.1, "MUS-2": 0.2 };

export function calculateFee(dailyList) {
  return dailyList.map((row) => {
    const record = {
      province: row.PROVINCE.trim(),
      weight: Number(row.WEIGHT),
      customer: row.CUSTOMER.trim(),
      count: Number(row.COUNT),
    };
    record.zone = ZONE[record.province] ?? "far";
    record.tier = record.weight <= 1 ? 1 : record.weight <= 5 ? 2 : record.weight <= 20 ? 3 : 4;
    record.raw = Math.max(MINIMUM[record.zone], 25 * record.tier * COEFFICIENT[record.zone]);
    record.net = record.raw * (1 - (CONTRACT[record.customer] ?? 0)) * (record.count >= 10 ? 0.95 : 1);
    record.net = Math.round(record.net * 100) / 100;
    return record;
  });
}

Filters

In the second arrangement, each step is its own file, and all carry the same signature: take a record, give a record. That the exported name is the same (filter) is the style’s condition — wherever the chain is built does not distinguish between steps.

// pipe/normalize.mjs — converts raw row field names into record fields
export const filter = (row) => ({
  province: row.PROVINCE.trim(),
  weight: Number(row.WEIGHT),
  customer: row.CUSTOMER.trim(),
  count: Number(row.COUNT),
});

// pipe/zone.mjs — derives the fee zone from the province
const ZONE = { "34": "near", "06": "mid", "35": "mid", "65": "far" };
export const filter = (r) => ({ ...r, zone: ZONE[r.province] ?? "far" });

// pipe/tariff.mjs — weight tier, zone coefficient and minimum fee
const COEFFICIENT = { near: 1, mid: 1.4, far: 2.1 };
const MINIMUM = { near: 30, mid: 45, far: 70 };
const tier = (kg) => (kg <= 1 ? 1 : kg <= 5 ? 2 : kg <= 20 ? 3 : 4);
export const filter = (r) => {
  const t = tier(r.weight);
  return { ...r, tier: t, raw: Math.max(MINIMUM[r.zone], 25 * t * COEFFICIENT[r.zone]) };
};

// pipe/discount.mjs — contracted customer discount and volume discount
const CONTRACT = { "MUS-1": 0.1, "MUS-2": 0.2 };
export const filter = (r) => ({
  ...r,
  net: r.raw * (1 - (CONTRACT[r.customer] ?? 0)) * (r.count >= 10 ? 0.95 : 1),
});

// pipe/round.mjs — rounds the amount to the smallest currency unit
export const filter = (r) => ({ ...r, net: Math.round(r.net * 100) / 100 });

Each filter carries forward what arrives with ...r and adds its own field on top; it never names a field it has no interest in.

The Chain Itself

The order sits in a single file as a list of names; the same file carries the generator that streams records one at a time. The watch parameter is for measurement — called at every filter’s entry.

// pipe/chain.mjs — the chain's order lives only here, no field name appears
export const CHAIN = ["normalize", "zone", "tariff", "discount", "round"];

export async function* stream(source, watch = () => {}) {
  const filters = [];
  for (const name of CHAIN) filters.push({ name, filter: (await import(`./${name}.mjs`)).filter });
  for await (const row of source) {
    let r = row;
    for (const { name, filter } of filters) {
      watch(name, r);
      r = filter(r);
    }
    yield r;
  }
}

Measurement

The script runs a six-row daily list through both arrangements, verifies the outputs match, and produces three numbers per step. Input and output field counts are counted at run time; the read count is names appearing in the file’s source as .fieldName.

// count-fields.mjs — both arrangements produce the same list, the fields each step reads and the fields crossing the boundary
import { readFileSync } from "node:fs";
import { calculateFee } from "./single/body.mjs";
import { CHAIN, stream } from "./pipe/chain.mjs";

const DAILY_LIST = [
  { PROVINCE: "34", WEIGHT: "0.8", CUSTOMER: "MUS-1", COUNT: "3" },
  { PROVINCE: "65", WEIGHT: "12", CUSTOMER: "MUS-2", COUNT: "12" },
  { PROVINCE: " 06", WEIGHT: "30", CUSTOMER: "MUS-9 ", COUNT: "1" },
  { PROVINCE: "35", WEIGHT: "4", CUSTOMER: "MUS-1", COUNT: "20" },
  { PROVINCE: "34", WEIGHT: "21", CUSTOMER: "MUS-2", COUNT: "2" },
  { PROVINCE: "65", WEIGHT: "0.4", CUSTOMER: "MUS-9", COUNT: "1" },
];
const FIELDS = ["PROVINCE", "WEIGHT", "CUSTOMER", "COUNT", "province", "weight", "customer", "count", "zone", "tier", "raw", "net"];
const fieldsRead = (path) => {
  const text = readFileSync(path, "utf8");
  return FIELDS.filter((f) => new RegExp(`\\.${f}\\b`).test(text)).length;
};

const inputFields = new Map();
const watch = (name, r) => { if (inputFields.has(name) === false) inputFields.set(name, Object.keys(r).length); };

const body = calculateFee(DAILY_LIST);
const chain = [];
for await (const r of stream(DAILY_LIST, watch)) chain.push(r);

console.log(`first record: ${JSON.stringify(chain[0])}`);
console.log(`both arrangements equal = ${JSON.stringify(body) === JSON.stringify(chain)}, records = ${chain.length}`);

console.log("\nstep        input field    fields read  output field");
let passed = 0, totalRead = 0;
for (const [i, name] of CHAIN.entries()) {
  const input = inputFields.get(name);
  const output = i + 1 < CHAIN.length ? inputFields.get(CHAIN[i + 1]) : Object.keys(chain[0]).length;
  const read = fieldsRead(`pipe/${name}.mjs`);
  passed += input;
  totalRead += read;
  console.log(
    `${name.padEnd(12)}${String(input).padStart(10)}${String(read).padStart(16)}${String(output).padStart(12)}`,
  );
}
console.log(
  `${"single body".padEnd(12)}${String(Object.keys(DAILY_LIST[0]).length).padStart(10)}` +
    `${String(fieldsRead("single/body.mjs")).padStart(16)}${String(Object.keys(body[0]).length).padStart(12)}`,
);
console.log(`\nfields crossing a filter boundary = ${passed}, of these read = ${totalRead}`);

node count-fields.mjs

first record: {"province":"34","weight":0.8,"customer":"MUS-1","count":3,"zone":"near","tier":1,"raw":30,"net":27}
both arrangements equal = true, records = 6

step        input field    fields read  output field
normalize            4               4           4
zone                 4               1           5
tariff               5               2           7
discount             7               3           8
round                8               1           8
single body          4              12           8

fields crossing a filter boundary = 28, of these read = 11

Reading the Numbers

Six records produced identical output in both arrangements, so the measurement compares two arrangements of the same work.

The fields-read column shows the distribution of the knowledge obligation. The single body recognizes all twelve names — four raw headers, four normalized, four derived. In the chain, the same information spreads across five files, none recognizing more than three; finding which file to edit when a field’s name changes is a matter of reading this column.

The input and output columns give the style’s cost. 28 fields crossed filter boundaries, only 11 read by name; the remaining 17 were carried without ever being read — each filter passes along what does not concern it, since it cannot know what the next step needs. The single body carries nothing: intermediate values sit in the same scope, no boundary crossed. The trade-off: spreading names out is paid for in carried fields.

A New Step

The scenario question is now tested. The fuel surcharge step enters after the discount and before the rounding. The script copies both arrangements, applies the change, and counts edited files. Its last section runs each filter standalone with the record captured at its own boundary, comparing the output against the record the next step received.

// extend.mjs — edited file, changed lines and that file's fields read when the fuel surcharge step is added
import { cpSync, readdirSync, readFileSync, writeFileSync } from "node:fs";

const FIELDS = ["PROVINCE", "WEIGHT", "CUSTOMER", "COUNT", "province", "weight", "customer", "count", "zone", "tier", "raw", "net"];
const fieldsRead = (text) => FIELDS.filter((f) => new RegExp(`\\.${f}\\b`).test(text)).length;
const DEFINITION = "const FUEL = { near: 0.02, mid: 0.04, far: 0.07 };";
const edit = (path, transform) => writeFileSync(path, transform(readFileSync(path, "utf8")));

for (const dir of ["single", "pipe"]) cpSync(dir, `new/${dir}`, { recursive: true });
edit("new/single/body.mjs", (m) =>
  m.replace("const CONTRACT", `${DEFINITION}\nconst CONTRACT`)
    .replace("    record.net = Math.round", "    record.net = record.net * (1 + FUEL[record.zone]);\n    record.net = Math.round"));
writeFileSync("new/pipe/fuel.mjs",
  `// pipe/fuel.mjs — fuel surcharge by zone\n${DEFINITION}\nexport const filter = (r) => ({ ...r, net: r.net * (1 + FUEL[r.zone]) });\n`);
edit("new/pipe/chain.mjs", (m) => m.replace('"discount", "round"', '"discount", "fuel", "round"'));

const changedLines = (previous, updated) => {
  const tally = (m) => m.split("\n").reduce((h, s) => h.set(s, (h.get(s) ?? 0) + 1), new Map());
  const a = tally(previous), b = tally(updated);
  const missing = (x, y) => [...x].reduce((t, [s, n]) => t + Math.max(0, n - (y.get(s) ?? 0)), 0);
  return missing(b, a) + missing(a, b);
};

console.log("arrangement  edited  added  changed lines  fields read in edited");
for (const dir of ["single", "pipe"]) {
  const existingFiles = new Set(readdirSync(dir));
  const edited = [];
  let added = 0, lines = 0, read = 0;
  for (const file of readdirSync(`new/${dir}`)) {
    const updatedText = readFileSync(`new/${dir}/${file}`, "utf8");
    if (existingFiles.has(file) === false) { added += 1; continue; }
    const previousText = readFileSync(`${dir}/${file}`, "utf8");
    if (previousText === updatedText) continue;
    edited.push(file);
    lines += changedLines(previousText, updatedText);
    read += fieldsRead(updatedText);
  }
  console.log(
    `${dir.padEnd(7)}${String(edited.length).padStart(10)}${String(added).padStart(9)}` +
      `${String(lines).padStart(15)}${String(read).padStart(25)}  [${edited.join(" ")}]`,
  );
}

const DAILY_LIST = [{ PROVINCE: "34", WEIGHT: "0.8", CUSTOMER: "MUS-1", COUNT: "3" }, { PROVINCE: "65", WEIGHT: "12", CUSTOMER: "MUS-2", COUNT: "12" }];
const { calculateFee: previous } = await import("./single/body.mjs");
const { calculateFee } = await import("./new/single/body.mjs");
const { CHAIN, stream } = await import("./new/pipe/chain.mjs");
const boundary = new Map();
const chain = [];
for await (const r of stream(DAILY_LIST, (name, r) => { if (boundary.has(name) === false) boundary.set(name, r); })) chain.push(r);
console.log(`\nsteps after new step = ${CHAIN.length}, both arrangements equal = ${JSON.stringify(calculateFee(DAILY_LIST)) === JSON.stringify(chain)}`);
console.log(`second record net: before fuel step ${previous(DAILY_LIST)[1].net}, after ${chain[1].net}`);

let tested = 0;
for (const [i, name] of CHAIN.entries()) {
  const { filter } = await import(`./new/pipe/${name}.mjs`);
  const expected = i + 1 < CHAIN.length ? boundary.get(CHAIN[i + 1]) : chain[0];
  if (JSON.stringify(filter(boundary.get(name))) === JSON.stringify(expected)) tested += 1;
}
console.log(`step run standalone with its own boundary record: pipeline ${tested}/${CHAIN.length}, single body 1/1`);

node extend.mjs

arrangement  edited  added  changed lines  fields read in edited
single          1        0              2                       12  [body.mjs]
pipe            1        1              2                        0  [chain.mjs]

steps after new step = 6, both arrangements equal = true
second record net: before fuel step 119.7, after 128.08
step run standalone with its own boundary record: pipeline 6/6, single body 1/1

The edited-file count is 1 in both arrangements; no gain there. The gain is in the last column: the single body’s edited file recognizes twelve field names, so the fuel surcharge means opening the body where tariff and discount sit. The chain’s edited file, chain.mjs, recognizes no field name at all — the change inserts a string into an array; the new step sits in a separate file, recognizing two names.

The last line gives the second gain. All six steps, called standalone with the record captured at their own boundary, produced the record the next step received. No such arrangement is possible in the single body: the only callable unit is the entire body, and intermediate values never step outside it. Testing the rounding rule requires running the tariff and discount too.

Streaming Records

The filter boundary being a single record has a side effect: the whole chain can run without waiting for the list. The script below runs both arrangements at three sizes and counts the highest number of records taken from the source whose output is not yet written — the record count, not a memory byte.

// streaming.mjs — records simultaneously live: when the list is taken in bulk and when it is streamed one by one
import { calculateFee } from "./single/body.mjs";
import { stream } from "./pipe/chain.mjs";

const PROVINCE = ["34", "06", "35", "65"], CUSTOMER = ["MUS-1", "MUS-2", "MUS-9"];
const row = (i) => ({
  PROVINCE: PROVINCE[i % PROVINCE.length], WEIGHT: String(1 + (i % 25)),
  CUSTOMER: CUSTOMER[i % CUSTOMER.length], COUNT: String(1 + (i % 12)),
});

function counter() {
  let live = 0, peak = 0;
  return {
    peak: () => peak,
    take: (r) => { live += 1; peak = Math.max(peak, live); return r; },
    release: () => { live -= 1; },
  };
}

function batch(n) {
  const c = counter();
  const rows = Array.from({ length: n }, (_, i) => c.take(row(i)));
  let total = 0;
  for (const r of calculateFee(rows)) { total += r.net; c.release(); }
  return { peak: c.peak(), total: Math.round(total * 100) / 100 };
}

async function streamed(n) {
  const c = counter();
  function* source() { for (let i = 0; i < n; i += 1) yield c.take(row(i)); }
  let total = 0;
  for await (const r of stream(source())) { total += r.net; c.release(); }
  return { peak: c.peak(), total: Math.round(total * 100) / 100 };
}

console.log("records  batch peak  streamed peak  totals equal");
for (const n of [6, 60, 600]) {
  const a = batch(n), b = await streamed(n);
  console.log(`${String(n).padStart(5)}${String(a.peak).padStart(13)}${String(b.peak).padStart(15)}${String(a.total === b.total).padStart(16)}`);
}

node streaming.mjs

records  batch peak  streamed peak  totals equal
    6            6              1            true
   60           60              1            true
  600          600              1            true

In the batch arrangement, the live count grew with the list size: 6, 60, 600. In the streamed chain it stayed at 1 for all three sizes, and both arrangements produced the same total. Because the number is independent of input size, the chain runs even if the list has no end; this is what stream means for the style.

The difference comes not from the style itself but from how the boundary is defined: the single body could also be called record by record, but it still keeps all five steps in one file, and those steps still cannot run one at a time. In the pipeline, streamability is a consequence of the filter signature, not an extra effort.

Summary

• In the pipeline style, each step is a filter: it takes a record, gives a record, and does not know its neighbor; order lives not in the filters but in the file that builds the chain.
• The six-record list produced the same result in both arrangements; the single body recognizes twelve field names, and none of the five filters recognizes more than three.
• 28 fields crossed filter boundaries, of which 11 were read; the remaining 17 were carried without ever being read — the cost of spreading the names out.
• Adding a new step edited 1 file in both arrangements, but the edited file recognizes 12 field names in the single body versus 0 in the chain.
• All six steps ran standalone with their own boundary record; the single body has 1 callable unit; when records were streamed, the live count stayed at 1 even for the 600-record list.

Next Step

The pipeline is a fixed, linear chain: every record passes through every step, and each step’s output is the next one’s input. The library holds work that does not fit this pattern. When a shipment’s fee is finalized, several things must happen: a record must be opened and a route assigned in the delivery operations context, the contracted customer’s volume counter must increase, and which tariff version priced it logged to the archive. These jobs do not feed each other, do not transform the record, and have no order among them; their count also changes over time. If the module that finalizes the fee calls them itself, it must import all of them, and every new job edits that file. The next lesson puts a named event and an event bus in between, then measures: the modules and names the publisher knows about the other side, its import closure, lines edited when a fourth consumer is added, and files that must be read to answer “who handles this event.”