Ροές

Προβολή ως Markdown Επεξεργασία στο GitHub Τεκμηρίωση στο 5b53f69

Το Phasis υλοποιεί ολόκληρο το πρότυπο WHATWG Streams. Κάθε κλάση, κάθε λειτουργία ανάγνωσης, κάθε βοηθητικό εργαλείο. Ο ίδιος κώδικας που λειτουργεί σε Node και σύγχρονα προγράμματα περιήγησης λειτουργεί και στο Phasis.

Η υλοποίηση είναι αυτή που χρησιμοποιεί εσωτερικά το fetch — το Response.body είναι πραγματικό ReadableStream, το Request δέχεται σώματα σε ροή και λειτουργεί το tee() που χρειάζεστε για να κλωνοποιήσετε ένα Response.

Τι περιλαμβάνεται

Κλάση	Κατάσταση
`ReadableStream`	Πλήρης
`ReadableStreamDefaultController`	Πλήρης
`ReadableByteStreamController`	Πλήρης
`ReadableStreamDefaultReader`	Πλήρης
`ReadableStreamBYOBReader`	Πλήρης
`ReadableStreamBYOBRequest`	Πλήρης
`WritableStream`	Πλήρης
`WritableStreamDefaultController`	Πλήρης
`WritableStreamDefaultWriter`	Πλήρης
`TransformStream`	Πλήρης
`TransformStreamDefaultController`	Πλήρης
`ByteLengthQueuingStrategy`	Πλήρης
`CountQueuingStrategy`	Πλήρης

Επιπλέον, οι μέθοδοι που χρησιμοποιείτε συχνότερα: pipeTo, pipeThrough, tee, getReader, ασύγχρονη επανάληψη, Symbol.asyncIterator, ανάγνωση BYOB με { min }.

Παραγωγή

JavaScript

// Pull-based — only generates values when read
const counter = new ReadableStream({
  start(controller) { this.n = 0; },
  pull(controller) {
    if (this.n >= 5) { controller.close(); return; }
    controller.enqueue(this.n++);
  },
});

for await (const v of counter) console.log(v);  // 0,1,2,3,4

JavaScript

// Push-based — start enqueues everything eagerly
const small = new ReadableStream({
  start(c) {
    [10, 20, 30].forEach(n => c.enqueue(n));
    c.close();
  },
});

Κατανάλωση

Τρεις τρόποι ανάγνωσης:

JavaScript

// 1. Reader
const reader = stream.getReader();
while (true) {
  const { value, done } = await reader.read();
  if (done) break;
  console.log(value);
}
reader.releaseLock();

// 2. Async iteration (preferred)
for await (const chunk of stream) {
  console.log(chunk);
}

// 3. tee + parallel pipes
const [a, b] = stream.tee();
await Promise.all([a.pipeTo(sinkA), b.pipeTo(sinkB)]);

Μετασχηματισμός

JavaScript

const upper = new TransformStream({
  transform(chunk, controller) {
    controller.enqueue(chunk.toUpperCase());
  },
});

const result = new ReadableStream({
  start(c) { ['hi', 'there'].forEach(s => c.enqueue(s)); c.close(); }
}).pipeThrough(upper);

for await (const v of result) console.log(v);  // HI, THERE

Ροές byte (BYOB)

JavaScript

const bytes = new ReadableStream({
  type: 'bytes',
  start(controller) {
    controller.enqueue(new Uint8Array([1, 2, 3, 4, 5, 6]));
    controller.close();
  },
});

const reader = bytes.getReader({ mode: 'byob' });
const buf = new Uint8Array(4);
const { value, done } = await reader.read(buf);
// value is a Uint8Array of length 4, sharing buffer with `buf`

Υποστηρίζεται το { min: N } — το read(view, { min: 2 }) ολοκληρώνεται μόνο όταν είναι διαθέσιμα τουλάχιστον 2 στοιχεία.

Έλεγχος οπισθοπίεσης

Τα desiredSize και ready λειτουργούν όπως περιγράφουν οι προδιαγραφές. Χρησιμοποιήστε στρατηγική ουράς για να δώσετε στον παραγωγό ένα κανάλι επιστροφής πληροφοριών:

JavaScript

const stream = new ReadableStream({
  pull(controller) {
    if (controller.desiredSize <= 0) return;  // back off
    controller.enqueue(nextChunk());
  },
}, new CountQueuingStrategy({ highWaterMark: 4 }));

Με fetch

JavaScript

// Response body is a ReadableStream
const res = await fetch('https://example.com/big.txt');
let total = 0;
for await (const chunk of res.body) {
  total += chunk.byteLength;
}

// Request bodies can stream too
const upload = new ReadableStream({
  start(c) { c.enqueue(new TextEncoder().encode('hello')); c.close(); }
});
await fetch('https://example.com/upload', { method: 'POST', body: upload });

Διαχωρισμός για προσωρινή αποθήκευση

Συνηθισμένο μοτίβο: διαβάστε το σώμα δύο φορές.

JavaScript

const res = await fetch(url);
const [forCache, forUse] = res.body.tee();
await Promise.all([
  forCache.pipeTo(cacheWritable),
  parseJson(forUse),
]);

Το Response.clone() το κάνει για εσάς και επιστρέφει ένα νέο Response.

Επισημάνσεις

Παραγωγοί μόνο σύγχρονης εκτέλεσης. Ένα pull() που εκτελεί αποκλειστικές κλήσεις εισόδου/εξόδου PHP μπλοκάρει το script. Για HTTP χρησιμοποιήστε το hook μεταφοράς fetch· για αρχεία, προτιμήστε να τα φορτώσετε πρώτα στη μνήμη. Διατίθενται τα setTimeout / setInterval (δείτε το Event loop), αλλά η μηχανή χρησιμοποιεί εικονικό χρόνο από προεπιλογή — ένα setTimeout(cb, 1000) μέσα σε pull() δεν προκαλεί αναμονή PHP, αλλά προωθεί το ρολόι του loop. Για ρυθμό πραγματικού χρόνου, ο host πρέπει να οδηγεί το tickEventLoop() με βάση ένα ρολόι τοίχου.

Δείτε επίσης

Web APIs (WPT) — πίνακας ποσοστού επιτυχίας WPT που καλύπτει τα υποτεστ Streams.
Μεταφορά Fetch — πώς διασχίζουν τα σώματα αιτημάτων και αποκρίσεων το όριο PHP/JS.
Μοτίβα AbortController — ακύρωση αναγνώσεων σε εξέλιξη.