Streams: Eksiksiz kılavuz

Streams API ile okunabilir, yazılabilir ve dönüştürülebilir akışları nasıl kullanacağınızı öğrenin.

Thomas Steiner

Streams API, ağ üzerinden alınan veya yerel olarak herhangi bir şekilde oluşturulan veri akışlarına programatik olarak erişmenize ve bunları JavaScript ile işlemenize olanak tanır. Akış, almak, göndermek veya dönüştürmek istediğiniz bir kaynağın küçük parçalara ayrılmasını ve bu parçaların bit bit işlenmesini içerir. Tarayıcılar, web sayfalarında gösterilecek HTML veya videolar gibi öğeleri alırken zaten akış gerçekleştirir. Ancak bu özellik, 2015'te fetch ile akışlar kullanıma sunulana kadar JavaScript'te hiç kullanılmamıştı.

Not akışı, teknik olarak XMLHttpRequest ile mümkündür ancak bu, ideal bir çözüm değildir. "XMLHttpRequest"in iyi bir örneğini gösteren bir GitHub Gist'i burada yer almaktadır.

Daha önce, bir kaynağı (video veya metin dosyası vb.) işlemek istediğinizde tüm dosyayı indirmeniz, uygun bir biçime seri durumdan çıkarılmasını beklemeniz ve ardından işlemeniz gerekiyordu. JavaScript'te akışların kullanıma sunulmasıyla birlikte bu durum değişiyor. Artık istemcide kullanılabilir hale gelir gelmez arabellek, dize veya blob oluşturmanıza gerek kalmadan ham verileri JavaScript ile aşamalı olarak işleyebilirsiniz. Bu, aşağıda listelediğim bazı kullanım alanlarının kilidini açar:

• Video efektleri: Okunabilir bir video akışını, efektleri gerçek zamanlı olarak uygulayan bir dönüştürme akışından geçirme.
• Veri (sıkıştırma) açma: Bir dosya akışını, seçici olarak (sıkıştırma) açan bir dönüştürme akışından geçirme.
• Resim kod çözme: Bir HTTP yanıtı akışını, baytların bit eşlem verilerine kodunu çözen bir dönüştürme akışından ve ardından bit eşlemleri PNG'lere çeviren başka bir dönüştürme akışından geçirme. Bir hizmet çalışanının fetch işleyicisine yüklenirse AVIF gibi yeni görüntü biçimlerini şeffaf bir şekilde çoklu doldurmanıza olanak tanır.

Tarayıcı desteği

ReadableStream ve WritableStream

TransformStream

Temel kavramlar

Çeşitli yayın türleri hakkında ayrıntılı bilgi vermeden önce bazı temel kavramları açıklayayım.

Topak

Parça, bir akışa yazılan veya akıştan okunan tek bir veri parçasıdır. Herhangi bir türde olabilir. Hatta akışlar farklı türlerdeki parçaları içerebilir. Çoğu zaman, bir parça belirli bir akış için en küçük veri birimi olmaz. Örneğin, bir bayt akışı tek baytlar yerine 16 KiB Uint8Array birimlerinden oluşan parçalar içerebilir.

Okunabilir akışlar

Okunabilir bir akış, okuyabileceğiniz bir veri kaynağını temsil eder. Başka bir deyişle, veriler okunabilir bir akıştan çıkar. Daha net bir ifadeyle, okunabilir akış, ReadableStream sınıfının bir örneğidir.

Yazılabilir akışlar

Yazılabilir akış, veri için yazabileceğiniz bir hedefi temsil eder. Diğer bir deyişle, veriler yazılabilir bir akışa girer. Daha net bir ifadeyle, yazılabilir akış, WritableStream sınıfının bir örneğidir.

Akışları dönüştürme

Dönüşüm akışı, bir çift akıştan oluşur: yazılabilir tarafı olarak bilinen yazılabilir bir akış ve okunabilir tarafı olarak bilinen okunabilir bir akış. Bunun gerçek dünyadaki metaforu, bir dilden diğerine anında çeviri yapan bir simultane tercüman olur. Dönüştürme akışına özgü bir şekilde, yazılabilir tarafa yazma işlemi yapıldığında okunabilir tarafta okunabilecek yeni veriler sağlanır. Dönüşüm akışı olarak writable ve readable özelliklerine sahip tüm nesneler kullanılabilir. Ancak standart TransformStream sınıfı, uygun şekilde iç içe geçmiş böyle bir çift oluşturmayı kolaylaştırır.

Boru zincirleri

Akışlar öncelikli olarak birbirlerine yönlendirilerek kullanılır. Okunabilir bir akış, okunabilir akışın pipeTo() yöntemi kullanılarak doğrudan yazılabilir bir akışa yönlendirilebilir veya okunabilir akışın pipeThrough() yöntemi kullanılarak önce bir veya daha fazla dönüştürme akışından yönlendirilebilir. Bu şekilde bir araya getirilmiş akışlar kümesi, boru zinciri olarak adlandırılır.

Geri basınç

Bir boru zinciri oluşturulduktan sonra, parçaların içinden ne kadar hızlı akması gerektiğiyle ilgili sinyaller yayar. Zincirdeki herhangi bir adım henüz parçaları kabul edemiyorsa boru zinciri boyunca geriye doğru bir sinyal yayar. Sonunda orijinal kaynağa parçaları bu kadar hızlı üretmeyi bırakması söylenir. Akışı normalleştirme işlemine geri basınç denir.

Teeing

Okunabilir bir akış, tee() yöntemi kullanılarak bölünebilir (büyük "T" harfinin şeklinden adını alır). Bu işlem, akışı kilitler (yani artık doğrudan kullanılamaz) ancak bağımsız olarak kullanılabilen iki yeni akış (dal) oluşturur. Yayınlar geri sarılamadığı veya yeniden başlatılamadığı için teeing de önemlidir. Bu konuyla ilgili daha fazla bilgiyi sonraki bölümlerde bulabilirsiniz.

Okunabilir bir akışın mekanikleri

Okunabilir akış, temel bir kaynaktan akan bir ReadableStream nesnesiyle JavaScript'te temsil edilen bir veri kaynağıdır. The ReadableStream() oluşturucu, belirtilen işleyicilerden okunabilir bir akış nesnesi oluşturup döndürür. İki tür temel kaynak vardır:

• Push kaynakları, eriştiğinizde sürekli olarak size veri gönderir. Akışa erişimi başlatmak, duraklatmak veya iptal etmek size bağlıdır. Canlı video akışları, sunucu tarafından gönderilen etkinlikler veya WebSocket'ler buna örnek olarak verilebilir.
• Çekme kaynakları, bağlandıktan sonra verileri açıkça istemenizi gerektirir. fetch() veya XMLHttpRequest çağrıları aracılığıyla yapılan HTTP işlemleri buna örnek verilebilir.

Akış verileri, parçalar adı verilen küçük birimler halinde sırayla okunur. Bir akışa yerleştirilen parçaların enqueued söylenir. Bu, okunmaya hazır bir şekilde sıraya alındıkları anlamına gelir. Dahili kuyruk, henüz okunmamış parçaları takip eder.

Sıraya alma stratejisi, bir akışın dahili sırasının durumuna göre nasıl geri basınç sinyali vermesi gerektiğini belirleyen bir nesnedir. Kuyruğa alma stratejisi, her parçaya bir boyut atar ve kuyruktaki tüm parçaların toplam boyutunu, yüksek su işareti olarak bilinen belirtilen bir sayıyla karşılaştırır.

Akışın içindeki parçalar bir okuyucu tarafından okunur. Bu okuyucu, verileri tek tek alır ve üzerinde istediğiniz işlemi yapmanıza olanak tanır. Okuyucu ve beraberindeki diğer işleme kodu tüketici olarak adlandırılır.

Bu bağlamdaki bir sonraki yapıya denetleyici adı verilir. Okunabilir her akışın, adından da anlaşılacağı gibi akışı kontrol etmenize olanak tanıyan ilişkili bir denetleyicisi vardır.

Aynı anda yalnızca bir okuyucu bir akışı okuyabilir. Bir okuyucu oluşturulup bir akışı okumaya başladığında (yani etkin okuyucu olduğunda) akışa kilitlenir. Akışınızı başka bir okuyucunun okumasını istiyorsanız genellikle başka bir işlem yapmadan önce ilk okuyucuyu serbest bırakmanız gerekir (ancak akışları tee edebilirsiniz).

Okunabilir bir akış oluşturma

Oluşturucusunu ReadableStream() çağırarak okunabilir bir akış oluşturursunuz. Oluşturucunun, oluşturulan akış örneğinin nasıl davranacağını tanımlayan yöntem ve özelliklere sahip bir nesneyi temsil eden isteğe bağlı bir bağımsız değişkeni underlyingSource vardır.

underlyingSource

Bu, aşağıdaki isteğe bağlı, geliştirici tanımlı yöntemleri kullanabilir:

• start(controller): Nesne oluşturulduğunda hemen çağrılır. Yöntem, akış kaynağına erişebilir ve akış işlevini ayarlamak için gereken diğer tüm işlemleri yapabilir. Bu işlem eşzamansız olarak yapılacaksa yöntem, başarıyı veya başarısızlığı bildirmek için bir söz döndürebilir. Bu yönteme iletilen controller parametresi ReadableStreamDefaultController.
• pull(controller): Daha fazla parça getirildikçe akışı kontrol etmek için kullanılabilir. Akışın dahili parça sırası dolana kadar veya sıra yüksek su seviyesine ulaşana kadar tekrar tekrar çağrılır. pull() çağrısının sonucu bir söz ise bu söz yerine getirilene kadar pull() tekrar çağrılmaz. Söz reddedilirse akış hatalı hale gelir.
• cancel(reason): Akış tüketicisi akışı iptal ettiğinde çağrılır.

const readableStream = new ReadableStream({
  start(controller) {
    /* … */
  },

  pull(controller) {
    /* … */
  },

  cancel(reason) {
    /* … */
  },
});

ReadableStreamDefaultController aşağıdaki yöntemleri destekler:

• ReadableStreamDefaultController.close() ilişkili akışı kapatır.
• ReadableStreamDefaultController.enqueue() Belirli bir parçayı ilişkili akışta sıraya alır.
• ReadableStreamDefaultController.error(), ilişkili akışla gelecekteki tüm etkileşimlerin hata vermesine neden olur.

/* … */
start(controller) {
  controller.enqueue('The first chunk!');
},
/* … */

queuingStrategy

ReadableStream() oluşturucusunun ikinci ve isteğe bağlı olan bağımsız değişkeni queuingStrategy'dir. Bu, akış için isteğe bağlı olarak bir kuyruğa alma stratejisi tanımlayan ve iki parametre alan bir nesnedir:

• highWaterMark: Bu sıralama stratejisi kullanılarak akışın en yüksek noktasını gösteren negatif olmayan bir sayı.
• size(chunk): Belirtilen parça değerinin sonlu ve negatif olmayan boyutunu hesaplayıp döndüren bir işlev. Sonuç, uygun ReadableStreamDefaultController.desiredSize özelliği aracılığıyla kendini gösteren geri basıncı belirlemek için kullanılır. Ayrıca temel kaynağın pull() yönteminin ne zaman çağrılacağını da yönetir.

const readableStream = new ReadableStream({
    /* … */
  },
  {
    highWaterMark: 10,
    size(chunk) {
      return chunk.length;
    },
  },
);

getReader() ve read() yöntemleri

Okunabilir bir akıştan okumak için ReadableStreamDefaultReader olan bir okuyucuya ihtiyacınız vardır. ReadableStream arayüzünün getReader() yöntemi bir okuyucu oluşturur ve akışı bu okuyucuya kilitler. Akış kilitliyken bu okuyucu serbest bırakılana kadar başka okuyucu alınamaz.

ReadableStreamDefaultReader arayüzünün read() yöntemi, akışın dahili kuyruğundaki sonraki parçaya erişim sağlayan bir söz döndürür. Akışın durumuna bağlı olarak bir sonuçla yerine getirir veya reddeder. Farklı olasılıklar şunlardır:

• Bir parça varsa söz,
  { value: chunk, done: false } biçiminde bir nesneyle yerine getirilir.
• Akış kapatılırsa söz,
  { value: undefined, done: true } biçiminde bir nesneyle yerine getirilir.
• Akışta hata oluşursa söz, ilgili hatayla reddedilir.

const reader = readableStream.getReader();
while (true) {
  const { done, value } = await reader.read();
  if (done) {
    console.log('The stream is done.');
    break;
  }
  console.log('Just read a chunk:', value);
}

locked özelliği

Okunabilir bir akışın kilitli olup olmadığını kontrol etmek için akışın ReadableStream.locked özelliğine erişebilirsiniz.

const locked = readableStream.locked;
console.log(`The stream is ${locked ? 'indeed' : 'not'} locked.`);

Okunabilir akış kod örnekleri

Aşağıdaki kod örneğinde tüm adımlar gösterilmektedir. Öncelikle, ReadableStream bağımsız değişkeninde (yani TimestampSource sınıfında) start() yöntemini tanımlayan bir underlyingSource oluşturun. Bu yöntem, akışın controller değerine enqueue() on saniye boyunca her saniyede bir zaman damgası eklemesini söyler. Son olarak, kumandaya akışı close() talimatını verir. Bu akışı, getReader() yöntemiyle bir okuyucu oluşturup akış done olana kadar read() yöntemini çağırarak tüketirsiniz.

class TimestampSource {
  #interval

  start(controller) {
    this.#interval = setInterval(() => {
      const string = new Date().toLocaleTimeString();
      // Add the string to the stream.
      controller.enqueue(string);
      console.log(`Enqueued ${string}`);
    }, 1_000);

    setTimeout(() => {
      clearInterval(this.#interval);
      // Close the stream after 10s.
      controller.close();
    }, 10_000);
  }

  cancel() {
    // This is called if the reader cancels.
    clearInterval(this.#interval);
  }
}

const stream = new ReadableStream(new TimestampSource());

async function concatStringStream(stream) {
  let result = '';
  const reader = stream.getReader();
  while (true) {
    // The `read()` method returns a promise that
    // resolves when a value has been received.
    const { done, value } = await reader.read();
    // Result objects contain two properties:
    // `done`  - `true` if the stream has already given you all its data.
    // `value` - Some data. Always `undefined` when `done` is `true`.
    if (done) return result;
    result += value;
    console.log(`Read ${result.length} characters so far`);
    console.log(`Most recently read chunk: ${value}`);
  }
}
concatStringStream(stream).then((result) => console.log('Stream complete', result));

Eşzamansız yineleme

Akışın done olup olmadığını her read() döngü yinelemesinde kontrol etmek en uygun API olmayabilir. Neyse ki yakında bunu yapmanın daha iyi bir yolu olacak: eşzamansız yineleme.

for await (const chunk of stream) {
  console.log(chunk);
}

Asenkron yinelemeyi kullanmak için geçici çözüm olarak, davranışı bir polyfill ile uygulayabilirsiniz.

if (!ReadableStream.prototype[Symbol.asyncIterator]) {
  ReadableStream.prototype[Symbol.asyncIterator] = async function* () {
    const reader = this.getReader();
    try {
      while (true) {
        const {done, value} = await reader.read();
        if (done) {
          return;
          }
        yield value;
      }
    }
    finally {
      reader.releaseLock();
    }
  }
}

Okunabilir bir akışı hazırlama

tee() yöntemi, ReadableStream arayüzü, mevcut okunabilir akışı hazırlar ve sonuçtaki iki dalı yeni ReadableStream örnekleri olarak içeren iki öğeli bir dizi döndürür. Bu sayede iki okuyucu aynı anda bir akışı okuyabilir. Örneğin, sunucudan yanıt getirip tarayıcıya aktarmak ve aynı zamanda hizmet çalışanı önbelleğine aktarmak istiyorsanız bunu bir hizmet çalışanında yapabilirsiniz. Yanıt gövdesi birden fazla kez kullanılamadığından bunu yapmak için iki kopya gerekir. Akışı iptal etmek için her iki sonucu da iptal etmeniz gerekir. Bir yayını başlatmak genellikle yayını süre boyunca kilitler ve diğer okuyucuların kilitlemesini engeller.

const readableStream = new ReadableStream({
  start(controller) {
    // Called by constructor.
    console.log('[start]');
    controller.enqueue('a');
    controller.enqueue('b');
    controller.enqueue('c');
  },
  pull(controller) {
    // Called `read()` when the controller's queue is empty.
    console.log('[pull]');
    controller.enqueue('d');
    controller.close();
  },
  cancel(reason) {
    // Called when the stream is canceled.
    console.log('[cancel]', reason);
  },
});

// Create two `ReadableStream`s.
const [streamA, streamB] = readableStream.tee();

// Read streamA iteratively one by one. Typically, you
// would not do it this way, but you certainly can.
const readerA = streamA.getReader();
console.log('[A]', await readerA.read()); //=> {value: "a", done: false}
console.log('[A]', await readerA.read()); //=> {value: "b", done: false}
console.log('[A]', await readerA.read()); //=> {value: "c", done: false}
console.log('[A]', await readerA.read()); //=> {value: "d", done: false}
console.log('[A]', await readerA.read()); //=> {value: undefined, done: true}

// Read streamB in a loop. This is the more common way
// to read data from the stream.
const readerB = streamB.getReader();
while (true) {
  const result = await readerB.read();
  if (result.done) break;
  console.log('[B]', result);
}

Okunabilir bayt akışları

Baytları temsil eden akışlar için, baytların verimli bir şekilde işlenmesi (özellikle kopyaların en aza indirilmesi) amacıyla okunabilir akışın genişletilmiş bir sürümü sağlanır. Bayt akışları, kendi arabelleğini getirme (BYOB) okuyucularının edinilmesine olanak tanır. Varsayılan uygulama, WebSockets durumunda dizeler veya dizi arabellekleri gibi çeşitli çıkışlar verebilirken bayt akışları bayt çıkışını garanti eder. Ayrıca BYOB okuyucular kararlılık avantajlarına sahiptir. Bunun nedeni, bir arabellek ayrılırsa aynı arabelleğe iki kez yazılmayacağının garanti edilebilmesi ve böylece yarış durumlarının önlenmesidir. BYOB okuyucular, arabellekleri yeniden kullanabildikleri için tarayıcının atık toplama işlemini çalıştırması gereken zaman sayısını azaltabilir.

Okunabilir bir bayt akışı oluşturma

type parametresini ReadableStream() oluşturucusuna ileterek okunabilir bir bayt akışı oluşturabilirsiniz.

new ReadableStream({ type: 'bytes' });

underlyingSource

Okunabilir bir bayt akışının temel kaynağı, üzerinde işlem yapmak için ReadableByteStreamController olarak verilir. ReadableByteStreamController.enqueue() yöntemi, değeri ArrayBufferView olan bir chunk bağımsız değişkeni alır. ReadableByteStreamController.byobRequest özelliği, mevcut BYOB çekme isteğini veya yoksa boş değeri döndürür. Son olarak, ReadableByteStreamController.desiredSize özelliği, kontrollü akışın dahili kuyruğunu doldurmak için istenen boyutu döndürür.

queuingStrategy

ReadableStream() oluşturucusunun ikinci ve isteğe bağlı olan bağımsız değişkeni queuingStrategy'dir. Bu, isteğe bağlı olarak akış için bir kuyruğa alma stratejisi tanımlayan ve tek bir parametre alan bir nesnedir:

• highWaterMark: Bu kuyruğa alma stratejisi kullanılarak akışın yüksek su işareti gösteren negatif olmayan bir bayt sayısı. Bu, uygun ReadableByteStreamController.desiredSize özelliği aracılığıyla kendini gösteren geri basıncı belirlemek için kullanılır. Ayrıca temel kaynağın pull() yönteminin ne zaman çağrılacağını da yönetir.

getReader() ve read() yöntemleri

Ardından, mode parametresini uygun şekilde ayarlayarak ReadableStreamBYOBReader erişebilirsiniz: ReadableStream.getReader({ mode: "byob" }). Bu sayede, kopyaları önlemek için arabellek ayırma işlemi üzerinde daha hassas bir kontrol sağlanır. Bayt akışından okumak için ReadableStreamBYOBReader.read(view) işlevini çağırmanız gerekir. Burada view, ArrayBufferView değeridir.

Okunabilir bayt akışı kod örneği

const reader = readableStream.getReader({ mode: "byob" });

let startingAB = new ArrayBuffer(1_024);
const buffer = await readInto(startingAB);
console.log("The first 1024 bytes, or less:", buffer);

async function readInto(buffer) {
  let offset = 0;

  while (offset < buffer.byteLength) {
    const { value: view, done } =
        await reader.read(new Uint8Array(buffer, offset, buffer.byteLength - offset));
    buffer = view.buffer;
    if (done) {
      break;
    }
    offset += view.byteLength;
  }

  return buffer;
}

Aşağıdaki işlev, rastgele oluşturulmuş bir dizinin verimli sıfır kopyalı okunmasına olanak tanıyan okunabilir bayt akışları döndürür. 1.024'lük önceden belirlenmiş bir parça boyutu kullanmak yerine, geliştirici tarafından sağlanan arabelleği doldurmaya çalışır ve tam kontrol sağlar.

const DEFAULT_CHUNK_SIZE = 1_024;

function makeReadableByteStream() {
  return new ReadableStream({
    type: 'bytes',

    pull(controller) {
      // Even when the consumer is using the default reader,
      // the auto-allocation feature allocates a buffer and
      // passes it to us via `byobRequest`.
      const view = controller.byobRequest.view;
      view = crypto.getRandomValues(view);
      controller.byobRequest.respond(view.byteLength);
    },

    autoAllocateChunkSize: DEFAULT_CHUNK_SIZE,
  });
}

Yazılabilir akışın mekanizması

Yazılabilir akış, veri yazabileceğiniz bir hedeftir ve JavaScript'te WritableStream nesnesiyle temsil edilir. Bu, temel bir alıcı üzerinde soyutlama görevi görür. Temel alıcı, ham verilerin yazıldığı daha düşük düzeyli bir G/Ç alıcısıdır.

Veriler, akışa yazıcı aracılığıyla tek tek parçalar halinde yazılır. Bir parça, okuyucudaki parçalar gibi çok çeşitli şekillerde olabilir. Yazmaya hazır parçalar oluşturmak için istediğiniz kodu kullanabilirsiniz. Yazar ve ilişkili koda üretici adı verilir.

Bir yazar oluşturulup bir akışa yazmaya başladığında (etkin yazar) bu akışa kilitlendiği söylenir. Yazılabilir bir akışa aynı anda yalnızca bir yazar yazabilir. Başka bir yazarın akışınıza yazmaya başlamasını istiyorsanız genellikle akışı yayınlamanız ve ardından başka bir yazarı eklemeniz gerekir.

Dahili kuyruk, akışa yazılan ancak henüz temel hedef tarafından işlenmemiş parçaları takip eder.

Sıraya alma stratejisi, bir akışın dahili sırasının durumuna göre nasıl geri basınç sinyali vermesi gerektiğini belirleyen bir nesnedir. Kuyruğa alma stratejisi, her parçaya bir boyut atar ve kuyruktaki tüm parçaların toplam boyutunu, yüksek su işareti olarak bilinen belirtilen bir sayıyla karşılaştırır.

Son yapıya denetleyici adı verilir. Yazılabilir her akışın, akışı kontrol etmenize (ör. akışı durdurmak için) olanak tanıyan ilişkili bir denetleyicisi vardır.

Yazılabilir akış oluşturma

Streams API'nin WritableStream arayüzü, akış verilerini hedef olarak bilinen bir alıcıya yazmak için standart bir soyutlama sağlar. Bu nesne, yerleşik geri basınç ve sıralama özellikleriyle birlikte gelir. Yazılabilir bir akış oluşturmak için yapılandırıcısını çağırın WritableStream(). Oluşturulan akış örneğinin nasıl davranacağını tanımlayan yöntem ve özelliklere sahip bir nesneyi temsil eden isteğe bağlı bir underlyingSink parametresi vardır.

underlyingSink

underlyingSink, aşağıdaki isteğe bağlı, geliştirici tanımlı yöntemleri içerebilir. Bazı yöntemlere iletilen controller parametresi bir WritableStreamDefaultController.

• start(controller): Bu yöntem, nesne oluşturulduğunda hemen çağrılır. Bu yöntemin içeriği, temel alttaki havuza erişmeyi amaçlamalıdır. Bu işlem eşzamansız olarak yapılacaksa başarıyı veya başarısızlığı bildirmek için bir söz döndürebilir.
• write(chunk, controller): Bu yöntem, yeni bir veri parçası (chunk parametresinde belirtilir) temel alttaki havuzda yazılmaya hazır olduğunda çağrılır. Yazma işleminin başarılı veya başarısız olduğunu bildirmek için bir söz döndürebilir. Bu yöntem yalnızca önceki yazma işlemleri başarılı olduktan sonra çağrılır ve akış kapatıldıktan veya iptal edildikten sonra asla çağrılmaz.
• close(controller): Bu yöntem, uygulama akışa parça yazmayı bitirdiğini işaret ederse çağrılır. İçerikler, temel alttaki hedefe yazma işlemlerini tamamlamak ve erişimi serbest bırakmak için gereken her şeyi yapmalıdır. Bu işlem eşzamansızsa başarıyı veya başarısızlığı bildirmek için bir söz döndürebilir. Bu yöntem yalnızca sıraya alınmış tüm yazma işlemleri başarılı olduktan sonra çağrılır.
• abort(reason): Bu yöntem, uygulama akışı aniden kapatmak ve hata durumuna geçirmek istediğini belirtirse çağrılır. close() işlevi gibi bekletilen kaynakları temizleyebilir ancak yazma işlemleri sıraya alınmış olsa bile abort() işlevi çağrılır. Bu parçalar atılır. Bu işlem eşzamansızsa başarıyı veya başarısızlığı bildirmek için bir söz döndürebilir. reason parametresi, akışın neden durdurulduğunu açıklayan bir DOMString içeriyor.

const writableStream = new WritableStream({
  start(controller) {
    /* … */
  },

  write(chunk, controller) {
    /* … */
  },

  close(controller) {
    /* … */
  },

  abort(reason) {
    /* … */
  },
});

Streams API'nin WritableStreamDefaultController arayüzü, kurulum sırasında, yazma için daha fazla parça gönderilirken veya yazma işleminin sonunda WritableStream durumunun kontrol edilmesine olanak tanıyan bir denetleyiciyi temsil eder. WritableStream oluşturulurken temel alıcıya, üzerinde işlem yapmak için karşılık gelen bir WritableStreamDefaultController örneği verilir. WritableStreamDefaultController yalnızca bir yönteme sahiptir: WritableStreamDefaultController.error(), bu da ilişkili yayınla gelecekteki tüm etkileşimlerin hata vermesine neden olur. WritableStreamDefaultController, AbortSignal örneğini döndüren bir signal özelliğini de destekler. Bu özellik, gerekirse WritableStream işleminin durdurulmasına olanak tanır.

/* … */
write(chunk, controller) {
  try {
    // Try to do something dangerous with `chunk`.
  } catch (error) {
    controller.error(error.message);
  }
},
/* … */

queuingStrategy

WritableStream() oluşturucusunun ikinci ve isteğe bağlı olan bağımsız değişkeni queuingStrategy'dir. Bu, akış için isteğe bağlı olarak bir kuyruğa alma stratejisi tanımlayan ve iki parametre alan bir nesnedir:

• highWaterMark: Bu sıralama stratejisi kullanılarak akışın en yüksek noktasını gösteren negatif olmayan bir sayı.
• size(chunk): Belirtilen parça değerinin sonlu ve negatif olmayan boyutunu hesaplayıp döndüren bir işlev. Sonuç, uygun WritableStreamDefaultWriter.desiredSize özelliği aracılığıyla kendini gösteren geri basıncı belirlemek için kullanılır.

getWriter() ve write() yöntemleri

Yazılabilir bir akışa yazmak için bir yazıcıya ihtiyacınız vardır. Yazıcı, WritableStreamDefaultWriter olacaktır. WritableStream arayüzünün getWriter() yöntemi, WritableStreamDefaultWriter öğesinin yeni bir örneğini döndürür ve akışı bu örnekle kilitler. Akış kilitliyken mevcut yazar serbest bırakılana kadar başka yazar alınamaz.

write() yöntemi, WritableStreamDefaultWriter arayüzü, iletilen bir veri parçasını WritableStream ve temelindeki hedefe yazar, ardından yazma işleminin başarılı veya başarısız olduğunu belirten bir söz döndürür. "Başarı"nın ne anlama geldiğinin temel alıcıya bağlı olduğunu unutmayın. Bu, parçanın kabul edildiğini gösterebilir ancak nihai hedefine güvenli bir şekilde kaydedildiğini göstermeyebilir.

const writer = writableStream.getWriter();
const resultPromise = writer.write('The first chunk!');

locked özelliği

Yazılabilir bir akışın kilitli olup olmadığını, WritableStream.locked özelliğine erişerek kontrol edebilirsiniz.

const locked = writableStream.locked;
console.log(`The stream is ${locked ? 'indeed' : 'not'} locked.`);

Yazılabilir akış kod örneği

Aşağıdaki kod örneğinde tüm adımlar gösterilmektedir.

const writableStream = new WritableStream({
  start(controller) {
    console.log('[start]');
  },
  async write(chunk, controller) {
    console.log('[write]', chunk);
    // Wait for next write.
    await new Promise((resolve) => setTimeout(() => {
      document.body.textContent += chunk;
      resolve();
    }, 1_000));
  },
  close(controller) {
    console.log('[close]');
  },
  abort(reason) {
    console.log('[abort]', reason);
  },
});

const writer = writableStream.getWriter();
const start = Date.now();
for (const char of 'abcdefghijklmnopqrstuvwxyz') {
  // Wait to add to the write queue.
  await writer.ready;
  console.log('[ready]', Date.now() - start, 'ms');
  // The Promise is resolved after the write finishes.
  writer.write(char);
}
await writer.close();

Okunabilir bir akışı yazılabilir bir akışa yönlendirme

Okunabilir bir akış, okunabilir akışın pipeTo() yöntemiyle yazılabilir bir akışa yönlendirilebilir. ReadableStream.pipeTo(), mevcut ReadableStream öğesini belirli bir WritableStream öğesine yönlendirir ve yönlendirme işlemi başarıyla tamamlandığında yerine getirilen veya herhangi bir hatayla karşılaşıldığında reddedilen bir söz döndürür.

const readableStream = new ReadableStream({
  start(controller) {
    // Called by constructor.
    console.log('[start readable]');
    controller.enqueue('a');
    controller.enqueue('b');
    controller.enqueue('c');
  },
  pull(controller) {
    // Called when controller's queue is empty.
    console.log('[pull]');
    controller.enqueue('d');
    controller.close();
  },
  cancel(reason) {
    // Called when the stream is canceled.
    console.log('[cancel]', reason);
  },
});

const writableStream = new WritableStream({
  start(controller) {
    // Called by constructor
    console.log('[start writable]');
  },
  async write(chunk, controller) {
    // Called upon writer.write()
    console.log('[write]', chunk);
    // Wait for next write.
    await new Promise((resolve) => setTimeout(() => {
      document.body.textContent += chunk;
      resolve();
    }, 1_000));
  },
  close(controller) {
    console.log('[close]');
  },
  abort(reason) {
    console.log('[abort]', reason);
  },
});

await readableStream.pipeTo(writableStream);
console.log('[finished]');

Dönüşüm akışı oluşturma

Streams API'nin TransformStream arayüzü, dönüştürülebilir bir veri grubunu temsil eder. Yapılandırıcısını TransformStream() çağırarak bir dönüştürme akışı oluşturursunuz. Bu, verilen işleyicilerden bir dönüştürme akışı nesnesi oluşturup döndürür. TransformStream() oluşturucusu, ilk bağımsız değişken olarak transformer öğesini temsil eden isteğe bağlı bir JavaScript nesnesini kabul eder. Bu tür nesneler aşağıdaki yöntemlerden herhangi birini içerebilir:

transformer

• start(controller): Bu yöntem, nesne oluşturulduğunda hemen çağrılır. Genellikle controller.enqueue() kullanılarak önek parçalarını sıraya almak için kullanılır. Bu parçalar okunabilir taraftan okunur ancak yazılabilir tarafa yapılan yazma işlemlerine bağlı değildir. Bu ilk işlem, örneğin önek parçalarını edinmek biraz çaba gerektirdiğinden eşzamansızsa işlev, başarıyı veya başarısızlığı belirtmek için bir söz döndürebilir. Reddedilen bir söz, akışta hataya neden olur. Oluşan tüm istisnalar, TransformStream() oluşturucusu tarafından yeniden oluşturulur.
• transform(chunk, controller): Bu yöntem, yazılabilir tarafa yazılan yeni bir parça dönüştürülmeye hazır olduğunda çağrılır. Akış uygulaması, bu işlevin yalnızca önceki dönüşümler başarılı olduktan sonra çağrılmasını ve hiçbir zaman start() tamamlanmadan veya flush() çağrıldıktan sonra çağrılmamasını sağlar. Bu işlev, dönüştürme akışının gerçek dönüştürme işlemini gerçekleştirir. Sonuçları controller.enqueue() kullanarak sıraya alabilir. Bu, controller.enqueue() kaç kez çağrıldığına bağlı olarak, yazılabilir tarafa yazılan tek bir parçanın okunabilir tarafta sıfır veya birden çok parçaya dönüşmesine olanak tanır. Dönüştürme işlemi eşzamansızsa bu işlev, dönüştürmenin başarılı veya başarısız olduğunu bildirmek için bir söz döndürebilir. Reddedilen bir söz, dönüştürme akışının hem okunabilir hem de yazılabilir taraflarında hataya neden olur. transform() yöntemi sağlanmazsa kimlik dönüşümü kullanılır. Bu dönüşüm, yazılabilir taraftaki değişmemiş parçaları okunabilir tarafa sıraya alır.
• flush(controller): Bu yöntem, yazılabilir tarafa yazılan tüm parçalar transform()'dan başarıyla geçerek dönüştürüldükten ve yazılabilir taraf kapatılmak üzereyken çağrılır. Bu genellikle, okunabilir taraf da kapanmadan önce sonek parçalarını okunabilir tarafa eklemek için kullanılır. Temizleme işlemi eşzamansızsa işlev, başarıyı veya başarısızlığı bildirmek için bir söz döndürebilir. Sonuç, stream.writable.write() işlevinin arayanına iletilir. Ayrıca, reddedilen bir söz, akışın hem okunabilir hem de yazılabilir tarafında hataya neden olur. İstisna oluşturma, reddedilen bir sözü döndürmeyle aynı şekilde değerlendirilir.

const transformStream = new TransformStream({
  start(controller) {
    /* … */
  },

  transform(chunk, controller) {
    /* … */
  },

  flush(controller) {
    /* … */
  },
});

writableStrategy ve readableStrategy sıralama stratejileri

TransformStream() oluşturucusunun ikinci ve üçüncü isteğe bağlı parametreleri writableStrategy ve readableStrategy kuyruğa alma stratejileridir. Bunlar sırasıyla okunabilir ve yazılabilir akış bölümlerinde belirtildiği şekilde tanımlanır.

Dönüştürme akışı kod örneği

Aşağıdaki kod örneğinde, dönüştürme akışının nasıl çalıştığı gösterilmektedir.

// Note that `TextEncoderStream` and `TextDecoderStream` exist now.
// This example shows how you would have done it before.
const textEncoderStream = new TransformStream({
  transform(chunk, controller) {
    console.log('[transform]', chunk);
    controller.enqueue(new TextEncoder().encode(chunk));
  },
  flush(controller) {
    console.log('[flush]');
    controller.terminate();
  },
});

(async () => {
  const readStream = textEncoderStream.readable;
  const writeStream = textEncoderStream.writable;

  const writer = writeStream.getWriter();
  for (const char of 'abc') {
    writer.write(char);
  }
  writer.close();

  const reader = readStream.getReader();
  for (let result = await reader.read(); !result.done; result = await reader.read()) {
    console.log('[value]', result.value);
  }
})();

Okunabilir bir akışı dönüştürme akışından geçirme

ReadableStream arayüzünün pipeThrough() yöntemi, mevcut akışı bir dönüştürme akışından veya başka bir yazılabilir/okunabilir çiftten geçirmek için zincirlenebilir bir yol sağlar. Bir akışın yönlendirilmesi genellikle yönlendirme süresi boyunca akışı kilitler ve diğer okuyucuların akışı kilitlemesini engeller.

const transformStream = new TransformStream({
  transform(chunk, controller) {
    console.log('[transform]', chunk);
    controller.enqueue(new TextEncoder().encode(chunk));
  },
  flush(controller) {
    console.log('[flush]');
    controller.terminate();
  },
});

const readableStream = new ReadableStream({
  start(controller) {
    // called by constructor
    console.log('[start]');
    controller.enqueue('a');
    controller.enqueue('b');
    controller.enqueue('c');
  },
  pull(controller) {
    // called read when controller's queue is empty
    console.log('[pull]');
    controller.enqueue('d');
    controller.close(); // or controller.error();
  },
  cancel(reason) {
    // called when rs.cancel(reason)
    console.log('[cancel]', reason);
  },
});

(async () => {
  const reader = readableStream.pipeThrough(transformStream).getReader();
  for (let result = await reader.read(); !result.done; result = await reader.read()) {
    console.log('[value]', result.value);
  }
})();

Biraz yapay olan sonraki kod örneğinde, döndürülen yanıt sözünü fetch() akış olarak tüketip metni parça parça büyük harfe çevirerek tüm metni büyük harfe çeviren bir "bağırma" sürümünün nasıl uygulanabileceği gösterilmektedir. Bu yaklaşımın avantajı, tüm belgenin indirilmesini beklemenize gerek olmamasıdır. Bu durum, büyük dosyalarla çalışırken büyük bir fark yaratabilir.

function upperCaseStream() {
  return new TransformStream({
    transform(chunk, controller) {
      controller.enqueue(chunk.toUpperCase());
    },
  });
}

function appendToDOMStream(el) {
  return new WritableStream({
    write(chunk) {
      el.append(chunk);
    }
  });
}

fetch('./lorem-ipsum.txt').then((response) =>
  response.body
    .pipeThrough(new TextDecoderStream())
    .pipeThrough(upperCaseStream())
    .pipeTo(appendToDOMStream(document.body))
);

Demo

Aşağıdaki demoda, okunabilir, yazılabilir ve dönüştürülebilir akışların nasıl çalıştığı gösterilmektedir. Ayrıca pipeThrough() ve pipeTo() boru zincirlerine örnekler de içerir ve tee()'ı gösterir. İsteğe bağlı olarak demoyu kendi penceresinde çalıştırabilir veya kaynak kodunu görüntüleyebilirsiniz.

Tarayıcıda kullanılabilen faydalı yayınlar

Tarayıcıda yerleşik olarak bulunan birçok yararlı akış vardır. Blob'dan kolayca ReadableStream oluşturabilirsiniz. Blob arayüzünün stream() yöntemi, okunduğunda blob'da bulunan verileri döndüren bir ReadableStream döndürür. Ayrıca, File nesnesinin belirli bir Blob türü olduğunu ve blob'un kullanılabildiği her bağlamda kullanılabileceğini unutmayın.

const readableStream = new Blob(['hello world'], { type: 'text/plain' }).stream();

TextDecoder.decode() ve TextEncoder.encode() öğelerinin yayın varyantlarına sırasıyla TextDecoderStream ve TextEncoderStream denir.

const response = await fetch('https://streams.spec.whatwg.org/');
const decodedStream = response.body.pipeThrough(new TextDecoderStream());

CompressionStream ve DecompressionStream dönüştürme akışlarını kullanarak dosyaları kolayca sıkıştırabilir veya sıkıştırmayı kaldırabilirsiniz. Aşağıdaki kod örneğinde, Streams spesifikasyonunu nasıl indirebileceğiniz, doğrudan tarayıcıda nasıl sıkıştırabileceğiniz (gzip) ve sıkıştırılmış dosyayı doğrudan diske nasıl yazabileceğiniz gösterilmektedir.

const response = await fetch('https://streams.spec.whatwg.org/');
const readableStream = response.body;
const compressedStream = readableStream.pipeThrough(new CompressionStream('gzip'));

const fileHandle = await showSaveFilePicker();
const writableStream = await fileHandle.createWritable();
compressedStream.pipeTo(writableStream);

File System Access API'nin FileSystemWritableFileStream ve deneysel fetch() istek akışları, gerçek hayattaki yazılabilir akışlara örnek olarak verilebilir.

Serial API, hem okunabilir hem de yazılabilir akışları yoğun bir şekilde kullanır.

// Prompt user to select any serial port.
const port = await navigator.serial.requestPort();
// Wait for the serial port to open.
await port.open({ baudRate: 9_600 });
const reader = port.readable.getReader();

// Listen to data coming from the serial device.
while (true) {
  const { value, done } = await reader.read();
  if (done) {
    // Allow the serial port to be closed later.
    reader.releaseLock();
    break;
  }
  // value is a Uint8Array.
  console.log(value);
}

// Write to the serial port.
const writer = port.writable.getWriter();
const data = new Uint8Array([104, 101, 108, 108, 111]); // hello
await writer.write(data);
// Allow the serial port to be closed later.
writer.releaseLock();

Son olarak, WebSocketStream API, akışları WebSocket API ile entegre eder.

const wss = new WebSocketStream(WSS_URL);
const { readable, writable } = await wss.connection;
const reader = readable.getReader();
const writer = writable.getWriter();

while (true) {
  const { value, done } = await reader.read();
  if (done) {
    break;
  }
  const result = await process(value);
  await writer.write(result);
}

Faydalı kaynaklar

• Akışlar spesifikasyonu
• Eşlik eden demolar
• Streams polyfill
• 2016: Web akışlarının yılı
• Eşzamansız yineleyiciler ve oluşturucular
• Akış Görselleştirici

Teşekkür

Bu makale; Jake Archibald, François Beaufort, Sam Dutton, Mattias Buelens, Surma, Joe Medley ve Adam Rice tarafından incelenmiştir. Jake Archibald'ın blog yayınları, akışları anlamamda bana çok yardımcı oldu. Kod örneklerinden bazıları GitHub kullanıcısı @bellbind'in keşiflerinden esinlenmiştir ve metinlerin bazı bölümleri MDN Web Docs on Streams'e dayanmaktadır. Streams Standard'ın yazarları, bu spesifikasyonu yazma konusunda muazzam bir iş çıkardı.