Borular ve Süzgeçler

Önceki üç varyantta veri tek bir sıçrayışla sınırı geçiyordu: model sonucu bir çağrıda verildi, biçimlendirme bir yerde yapıldı. Ücretlendirme bağlamının asıl iş akışı ise böyle değil. Günlük gönderi listesi işlenirken her kayıt sırayla beş adımdan geçer: alan adları normalleştirilir, bölge türetilir, tarife uygulanır, indirim düşülür, tutar yuvarlanır. Bugün bu beş adım tek bir gövdenin içinde duruyor; yeni bir adım — yakıt ek ücreti — istendiğinde o gövde düzenlenmek zorunda ve adımların hiçbiri tek başına koşturulamıyor.

Boru hattı (pipeline) biçemi bu işi adımlara böler. Her adım bir süzgeçtir (filter): girdisini bir kayıt olarak alır, çıktısını bir kayıt olarak verir, komşularının kim olduğunu bilmez. Adımlar arasındaki bağlantı borudur (pipe); zincirin sırası süzgeçlerin içinde değil, zinciri kuran yerde durur. Kabuk Programlama kursundaki Boru Hattı dersinde kurulan süreç zinciri bu biçemin bir örneğidir: orada süzgeç bir süreç, boru bir bellek tamponuydu; burada süzgeç bir modül, boru bir işlev çağrısıdır. Kabuk sözdizimi oraya aittir.

Bağlandığı kalite niteliği sürdürülebilirliktir ve nitelik sorusu şudur: zincire yeni bir işlem adımı eklendiğinde kaç dosya düzenlenir ve düzenlenen dosya kaç alan adı tanır. Yanında iki ölçü daha alınır: süzgeç girişlerinden geçen alan sayısı ve başarım verimliliği ailesine giren canlı kayıt sayısı.

Beş Adım Tek Gövdede

Birinci düzen bugünkü durumdur: tek bir işlev listeyi alır ve her satır için beş adımı sırayla uygular. Adımlar arasındaki geçiş bir sınır değil, aynı gövdedeki bir sonraki deyimdir.

mkdir -p tek boru

// tek/govde.mjs — bes adim tek govdenin icinde: alan adlari, bolge, tarife, indirim, yuvarlama
const BOLGE = { "34": "yakin", "06": "orta", "35": "orta", "65": "uzak" };
const KATSAYI = { yakin: 1, orta: 1.4, uzak: 2.1 };
const ASGARI = { yakin: 30, orta: 45, uzak: 70 };
const SOZLESME = { "MUS-1": 0.1, "MUS-2": 0.2 };

export function ucretle(gunlukListe) {
  return gunlukListe.map((satir) => {
    const kayit = {
      il: satir.IL.trim(),
      agirlik: Number(satir.AGIRLIK),
      musteri: satir.MUSTERI.trim(),
      adet: Number(satir.ADET),
    };
    kayit.bolge = BOLGE[kayit.il] ?? "uzak";
    kayit.kademe = kayit.agirlik <= 1 ? 1 : kayit.agirlik <= 5 ? 2 : kayit.agirlik <= 20 ? 3 : 4;
    kayit.ham = Math.max(ASGARI[kayit.bolge], 25 * kayit.kademe * KATSAYI[kayit.bolge]);
    kayit.net = kayit.ham * (1 - (SOZLESME[kayit.musteri] ?? 0)) * (kayit.adet >= 10 ? 0.95 : 1);
    kayit.net = Math.round(kayit.net * 100) / 100;
    return kayit;
  });
}

Süzgeçler

İkinci düzende her adım kendi dosyasındadır ve hepsi aynı imzayı taşır: bir kayıt alır, bir kayıt verir. Dışa açtıkları adın aynı olması (suz) biçemin koşuludur: zinciri kuran yer adımları birbirinden ayırt etmez.

// boru/normalle.mjs — ham satirin alan adlarini kayit alanlarina cevirir
export const suz = (satir) => ({
  il: satir.IL.trim(),
  agirlik: Number(satir.AGIRLIK),
  musteri: satir.MUSTERI.trim(),
  adet: Number(satir.ADET),
});

// boru/bolge.mjs — ilden ucret bolgesi turetir
const BOLGE = { "34": "yakin", "06": "orta", "35": "orta", "65": "uzak" };
export const suz = (k) => ({ ...k, bolge: BOLGE[k.il] ?? "uzak" });

// boru/tarife.mjs — agirlik kademesi, bolge katsayisi ve asgari ucret
const KATSAYI = { yakin: 1, orta: 1.4, uzak: 2.1 };
const ASGARI = { yakin: 30, orta: 45, uzak: 70 };
const kademe = (kg) => (kg <= 1 ? 1 : kg <= 5 ? 2 : kg <= 20 ? 3 : 4);
export const suz = (k) => {
  const d = kademe(k.agirlik);
  return { ...k, kademe: d, ham: Math.max(ASGARI[k.bolge], 25 * d * KATSAYI[k.bolge]) };
};

// boru/indirim.mjs — sozlesmeli musteri indirimi ve hacim indirimi
const SOZLESME = { "MUS-1": 0.1, "MUS-2": 0.2 };
export const suz = (k) => ({
  ...k,
  net: k.ham * (1 - (SOZLESME[k.musteri] ?? 0)) * (k.adet >= 10 ? 0.95 : 1),
});

// boru/yuvarla.mjs — tutari kurusa yuvarlar
export const suz = (k) => ({ ...k, net: Math.round(k.net * 100) / 100 });

Her süzgeç kendisine geleni ...k ile taşır ve üzerine kendi alanını ekler; ilgilendirmediği alanların adını yazmaz.

Zincirin Kendisi

Sıra tek bir dosyada, adların listesi olarak durur; aynı dosya kayıtları tek tek akıtan üreteci de taşır. gozle parametresi ölçüm içindir, her süzgeç girişinde çağrılır.

// boru/hat.mjs — zincirin sirasi yalnizca burada durur, hicbir alan adi gecmez
export const HAT = ["normalle", "bolge", "tarife", "indirim", "yuvarla"];

export async function* akit(kaynak, gozle = () => {}) {
  const suzgecler = [];
  for (const ad of HAT) suzgecler.push({ ad, suz: (await import(`./${ad}.mjs`)).suz });
  for await (const satir of kaynak) {
    let k = satir;
    for (const { ad, suz } of suzgecler) {
      gozle(ad, k);
      k = suz(k);
    }
    yield k;
  }
}

Ölçüm

Betik altı satırlık günlük listeyi iki düzende koşturur, çıktıların aynı olduğunu doğrular ve her adım için üç sayı çıkarır. Girdi ile çıktı alan sayısı koşumda sayılır; okunan alan sayısı dosya kaynağında .alanAdi biçiminde geçen adların sayısıdır.

// alan-say.mjs — iki duzenin ayni listeyi uretmesi, adimlarin okunan alanlari ve sinirdan gecen alan
import { readFileSync } from "node:fs";
import { ucretle } from "./tek/govde.mjs";
import { HAT, akit } from "./boru/hat.mjs";

const GUNLUK = [
  { IL: "34", AGIRLIK: "0.8", MUSTERI: "MUS-1", ADET: "3" },
  { IL: "65", AGIRLIK: "12", MUSTERI: "MUS-2", ADET: "12" },
  { IL: " 06", AGIRLIK: "30", MUSTERI: "MUS-9 ", ADET: "1" },
  { IL: "35", AGIRLIK: "4", MUSTERI: "MUS-1", ADET: "20" },
  { IL: "34", AGIRLIK: "21", MUSTERI: "MUS-2", ADET: "2" },
  { IL: "65", AGIRLIK: "0.4", MUSTERI: "MUS-9", ADET: "1" },
];
const ALAN = ["IL", "AGIRLIK", "MUSTERI", "ADET", "il", "agirlik", "musteri", "adet", "bolge", "kademe", "ham", "net"];
const okunanAlan = (yol) => {
  const metin = readFileSync(yol, "utf8");
  return ALAN.filter((a) => new RegExp(`\\.${a}\\b`).test(metin)).length;
};

const girdiAlan = new Map();
const gozle = (ad, k) => { if (girdiAlan.has(ad) === false) girdiAlan.set(ad, Object.keys(k).length); };

const govde = ucretle(GUNLUK);
const hat = [];
for await (const k of akit(GUNLUK, gozle)) hat.push(k);

console.log(`ilk kayit: ${JSON.stringify(hat[0])}`);
console.log(`iki duzen esit mi = ${JSON.stringify(govde) === JSON.stringify(hat)}, kayit = ${hat.length}`);

console.log("\nadim      girdi alan     okunan alan  cikti alan");
let gecen = 0, okunanToplam = 0;
for (const [i, ad] of HAT.entries()) {
  const girdi = girdiAlan.get(ad);
  const cikti = i + 1 < HAT.length ? girdiAlan.get(HAT[i + 1]) : Object.keys(hat[0]).length;
  const okunan = okunanAlan(`boru/${ad}.mjs`);
  gecen += girdi;
  okunanToplam += okunan;
  console.log(
    `${ad.padEnd(10)}${String(girdi).padStart(10)}${String(okunan).padStart(16)}${String(cikti).padStart(12)}`,
  );
}
console.log(
  `${"tek govde".padEnd(10)}${String(Object.keys(GUNLUK[0]).length).padStart(10)}` +
    `${String(okunanAlan("tek/govde.mjs")).padStart(16)}${String(Object.keys(govde[0]).length).padStart(12)}`,
);
console.log(`\nsuzgec girisinden gecen alan = ${gecen}, bunlardan okunan = ${okunanToplam}`);

node alan-say.mjs

ilk kayit: {"il":"34","agirlik":0.8,"musteri":"MUS-1","adet":3,"bolge":"yakin","kademe":1,"ham":30,"net":27}
iki duzen esit mi = true, kayit = 6

adim      girdi alan     okunan alan  cikti alan
normalle           4               4           4
bolge              4               1           5
tarife             5               2           7
indirim            7               3           8
yuvarla            8               1           8
tek govde          4              12           8

suzgec girisinden gecen alan = 28, bunlardan okunan = 11

Sayıların Okunması

Altı kayıt iki düzende birebir aynı çıktı, yani ölçüm aynı işin iki düzenlenişini karşılaştırıyor.

Okunan alan sütunu bilme yükümlülüğünün dağılımını verir. Tek gövde on iki alan adının hepsini tanıyor: dört ham başlık, dört normalleşmiş alan ve dört türetilmiş alan. Zincirde aynı bilgi beş dosyaya dağıldı ve hiçbir dosya üçten fazla ad tanımıyor; bir alanın adı değiştiğinde düzenlenecek dosyayı bulmak bu sütunu okumaktır.

Girdi ve çıktı sütunları biçemin bedelini verir. Süzgeç girişlerinden 28 alan geçti, bunlardan yalnız 11’i adıyla okundu. Kalan 17 alan hiç okunmadan taşındı — her süzgeç kendisini ilgilendirmeyeni de bir sonrakine iletmek zorunda, çünkü sonraki adımın neye ihtiyaç duyduğunu bilmiyor. Tek gövdede bu taşıma yok: ara değerler aynı kapsamda duruyor, sınır geçilmiyor. Ödünleşim budur — adların dağılması, taşınan alanla ödenir.

Yeni Bir Adım

Nitelik sorusu şimdi sınanır. Yakıt ek ücreti adımı indirimden sonra, yuvarlamadan önce girer. Betik iki düzenin kopyasını çıkarır, değişikliği ikisine de uygular ve düzenlenen dosyaları sayar. Son bölümde her süzgeç kendi sınırında yakalanan kayıtla tek başına çağrılır, çıktısı bir sonraki adımın aldığı kayıtla karşılaştırılır.

// uzanti.mjs — yakit ek ucreti adimi eklendiginde duzenlenen dosya, degisen satir ve o dosyanin okudugu alan
import { cpSync, readdirSync, readFileSync, writeFileSync } from "node:fs";

const ALAN = ["IL", "AGIRLIK", "MUSTERI", "ADET", "il", "agirlik", "musteri", "adet", "bolge", "kademe", "ham", "net"];
const okunanAlan = (metin) => ALAN.filter((a) => new RegExp(`\\.${a}\\b`).test(metin)).length;
const TANIM = "const YAKIT = { yakin: 0.02, orta: 0.04, uzak: 0.07 };";
const duzenle = (yol, don) => writeFileSync(yol, don(readFileSync(yol, "utf8")));

for (const d of ["tek", "boru"]) cpSync(d, `yeni/${d}`, { recursive: true });
duzenle("yeni/tek/govde.mjs", (m) =>
  m.replace("const SOZLESME", `${TANIM}\nconst SOZLESME`)
    .replace("    kayit.net = Math.round", "    kayit.net = kayit.net * (1 + YAKIT[kayit.bolge]);\n    kayit.net = Math.round"));
writeFileSync("yeni/boru/yakit.mjs",
  `// boru/yakit.mjs — bolgeye gore yakit ek ucreti\n${TANIM}\nexport const suz = (k) => ({ ...k, net: k.net * (1 + YAKIT[k.bolge]) });\n`);
duzenle("yeni/boru/hat.mjs", (m) => m.replace('"indirim", "yuvarla"', '"indirim", "yakit", "yuvarla"'));

const degisenSatir = (eski, yeni) => {
  const say = (m) => m.split("\n").reduce((h, s) => h.set(s, (h.get(s) ?? 0) + 1), new Map());
  const a = say(eski), b = say(yeni);
  const eksik = (x, y) => [...x].reduce((t, [s, n]) => t + Math.max(0, n - (y.get(s) ?? 0)), 0);
  return eksik(b, a) + eksik(a, b);
};

console.log("duzen  duzenlenen  eklenen  degisen satir  duzenlenende okunan alan");
for (const d of ["tek", "boru"]) {
  const eskiler = new Set(readdirSync(d));
  const duzenlenen = [];
  let eklenen = 0, satir = 0, okunan = 0;
  for (const ad of readdirSync(`yeni/${d}`)) {
    const yeniMetin = readFileSync(`yeni/${d}/${ad}`, "utf8");
    if (eskiler.has(ad) === false) { eklenen += 1; continue; }
    const eskiMetin = readFileSync(`${d}/${ad}`, "utf8");
    if (eskiMetin === yeniMetin) continue;
    duzenlenen.push(ad);
    satir += degisenSatir(eskiMetin, yeniMetin);
    okunan += okunanAlan(yeniMetin);
  }
  console.log(
    `${d.padEnd(7)}${String(duzenlenen.length).padStart(10)}${String(eklenen).padStart(9)}` +
      `${String(satir).padStart(15)}${String(okunan).padStart(25)}  [${duzenlenen.join(" ")}]`,
  );
}

const GUNLUK = [{ IL: "34", AGIRLIK: "0.8", MUSTERI: "MUS-1", ADET: "3" }, { IL: "65", AGIRLIK: "12", MUSTERI: "MUS-2", ADET: "12" }];
const { ucretle: eski } = await import("./tek/govde.mjs");
const { ucretle } = await import("./yeni/tek/govde.mjs");
const { HAT, akit } = await import("./yeni/boru/hat.mjs");
const sinir = new Map();
const hat = [];
for await (const k of akit(GUNLUK, (ad, k) => { if (sinir.has(ad) === false) sinir.set(ad, k); })) hat.push(k);
console.log(`\nyeni adimdan sonra adim = ${HAT.length}, iki duzen esit mi = ${JSON.stringify(ucretle(GUNLUK)) === JSON.stringify(hat)}`);
console.log(`ikinci kayit net: yakit adimindan once ${eski(GUNLUK)[1].net}, sonra ${hat[1].net}`);

let sinanan = 0;
for (const [i, ad] of HAT.entries()) {
  const { suz } = await import(`./yeni/boru/${ad}.mjs`);
  const beklenen = i + 1 < HAT.length ? sinir.get(HAT[i + 1]) : hat[0];
  if (JSON.stringify(suz(sinir.get(ad))) === JSON.stringify(beklenen)) sinanan += 1;
}
console.log(`kendi sinir kaydiyla tek basina kosturulan adim: boru hatti ${sinanan}/${HAT.length}, tek govde 1/1`);

node uzanti.mjs

duzen  duzenlenen  eklenen  degisen satir  duzenlenende okunan alan
tek             1        0              2                       12  [govde.mjs]
boru            1        1              2                        0  [hat.mjs]

yeni adimdan sonra adim = 6, iki duzen esit mi = true
ikinci kayit net: yakit adimindan once 119.7, sonra 128.08
kendi sinir kaydiyla tek basina kosturulan adim: boru hatti 6/6, tek govde 1/1

Düzenlenen dosya sayısı iki düzende de 1’dir; biçem bu sütunda bir kazanç sağlamadı. Kazanç son sütunda: tek gövdede düzenlenen dosya on iki alan adı tanır, yani yakıt ek ücretini eklemek tarife ve indirim hesabının durduğu gövdeyi açmayı gerektirir. Zincirde düzenlenen dosya hat.mjs hiçbir alan adı tanımıyor, değişiklik bir dizinin içine bir dize eklemektir; yeni adım ayrı bir dosyada, iki alan adı tanıyarak durur.

Son satır ikinci kazancı verir. Altı adımın altısı, kendi sınırında yakalanmış kayıtla tek başına çağrıldığında bir sonraki adımın aldığı kaydı üretti. Tek gövdede böyle bir düzenek kurulamaz: çağrılabilir tek birim gövdenin tamamıdır, ara değerler dışarıya çıkmaz. Yuvarlama kuralını sınamak için tarifeyi ve indirimi de koşturmak gerekir.

Kayıtların Akıtılması

Süzgeç sınırının tek kayıt olması bir yan sonuç doğurur: zincirin tamamı liste beklemeden sürülebilir. Aşağıdaki betik iki düzeni üç boyda koşturur ve kaynaktan alınmış ama çıktısı henüz yazılmamış kayıtların en yüksek sayısını sayar. Sayılan şey kayıt adedidir; bellek baytı ölçülmemiştir.

// akis.mjs — ayni anda canli olan kayit sayisi: liste toptan alindiginda ve tek tek akitildiginda
import { ucretle } from "./tek/govde.mjs";
import { akit } from "./boru/hat.mjs";

const IL = ["34", "06", "35", "65"], MUSTERI = ["MUS-1", "MUS-2", "MUS-9"];
const satir = (i) => ({
  IL: IL[i % IL.length], AGIRLIK: String(1 + (i % 25)),
  MUSTERI: MUSTERI[i % MUSTERI.length], ADET: String(1 + (i % 12)),
});

function sayac() {
  let canli = 0, tepe = 0;
  return {
    tepe: () => tepe,
    al: (k) => { canli += 1; tepe = Math.max(tepe, canli); return k; },
    yaz: () => { canli -= 1; },
  };
}

function toptan(n) {
  const s = sayac();
  const gunluk = Array.from({ length: n }, (_, i) => s.al(satir(i)));
  let toplam = 0;
  for (const k of ucretle(gunluk)) { toplam += k.net; s.yaz(); }
  return { tepe: s.tepe(), toplam: Math.round(toplam * 100) / 100 };
}

async function akitarak(n) {
  const s = sayac();
  function* kaynak() { for (let i = 0; i < n; i += 1) yield s.al(satir(i)); }
  let toplam = 0;
  for await (const k of akit(kaynak())) { toplam += k.net; s.yaz(); }
  return { tepe: s.tepe(), toplam: Math.round(toplam * 100) / 100 };
}

console.log("kayit  toptan tepe  akitarak tepe  toplam esit mi");
for (const n of [6, 60, 600]) {
  const a = toptan(n), b = await akitarak(n);
  console.log(`${String(n).padStart(5)}${String(a.tepe).padStart(13)}${String(b.tepe).padStart(15)}${String(a.toplam === b.toplam).padStart(16)}`);
}

node akis.mjs

kayit  toptan tepe  akitarak tepe  toplam esit mi
    6            6              1            true
   60           60              1            true
  600          600              1            true

Toptan düzende canlı kayıt sayısı liste boyuyla büyüdü: 6, 60, 600. Akıtılan zincirde üç boyda da 1 kaldı ve iki düzenin ürettiği toplam tutar eşit. Sayı girdi boyundan bağımsız olduğu için listenin sonu olmasa da zincir çalışabilir; akış (stream) kavramının biçemdeki karşılığı budur.

Bu fark biçemin kendisinden değil sınırın tanımından geliyor: tek gövde de kayıt kayıt çağrılacak biçimde yazılabilir, ama o gövde beş adımı yine tek dosyada tutar ve adımları tek tek koşturulamaz. Boru hattında akıtılabilirlik ek bir çaba değil, süzgeç imzasının sonucudur.

Özet

• Boru hattı biçeminde her adım bir süzgeçtir: bir kayıt alır, bir kayıt verir, komşusunu tanımaz; sıra süzgeçlerin içinde değil, zinciri kuran dosyada durur.
• Altı kayıtlık liste iki düzende aynı sonucu verdi; tek gövde on iki alan adı tanıyor, beş süzgeçten hiçbiri üçten fazla tanımıyor.
• Süzgeç girişlerinden 28 alan geçti, bunlardan 11’i okundu; kalan 17 alan hiç okunmadan taşındı ve bu, adları dağıtmanın bedelidir.
• Yeni bir adım eklemek iki düzende de 1 dosya düzenletti, ama düzenlenen dosya tek gövdede 12 alan adı tanırken zincirde 0 tanıyor.
• Altı adımın altısı kendi sınır kaydıyla tek başına koşturuldu, tek gövdede çağrılabilir birim sayısı 1’dir; kayıtlar akıtıldığında canlı kayıt sayısı 600’lük listede de 1 kaldı.

Sonraki Adım

Boru hattı sabit ve doğrusal bir zincirdir: her kayıt her adımdan geçer ve her adımın çıktısı bir sonrakinin girdisidir. Kitaplıkta bu kalıba uymayan bir iş kümesi var. Bir gönderinin ücreti kesinleştiğinde birkaç şey olmalı: teslimat operasyonu bağlamında kayıt açılıp rota atanmalı, sözleşmeli müşterinin hacim sayacı artmalı, hangi tarife sürümüyle fiyatlandığı arşive geçmeli. Bu işler birbirini beslemez, kaydı dönüştürmez ve aralarında bir sıra yoktur; sayıları da zamanla değişir. Ücreti kesinleştiren modül bunları kendisi çağırırsa hepsini ithal eder ve her yeni iş o dosyayı düzenletir. Sonraki ders araya adlandırılmış bir olay ve bir veriyolu koyar, sonra ölçer: yayımcının karşı taraf hakkında tanıdığı modül ve ad sayısı, yayımcının ithal kapanışı, dördüncü bir tüketici eklendiğinde yayımcıda düzenlenen satır sayısı ve “bu olayı kim işliyor” sorusu için okunması gereken dosya sayısı.