Rust Programlama Dili

Akışlar (Streams): Sıralı Future’lar

Bu bölümün başlarında, “Mesaj İletimiyle İki Görev Arasında Veri Gönderme” başlığında asenkron kanalımızın alıcı tarafını nasıl kullandığımızı hatırlayın. Asenkron recv metodu zaman içinde bir öğe dizisi üretir. Bu, akış (stream) olarak bilinen çok daha genel bir desenin bir örneğidir. Pek çok kavram doğal olarak akış şeklinde temsil edilir: bir kuyrukta öğelerin hazır hâle gelmesi, tüm veri kümesi bilgisayarın belleğine sığmayacak kadar büyük olduğunda verilerin dosya sisteminden artımlı olarak parça parça çekilmesi veya zaman içinde ağ üzerinden veri gelmesi. Akışlar her adımda bir future ürettiği için onları diğer her türlü future ile birlikte kullanabilir ve ilginç şekillerde birleştirebiliriz. Örneğin, çok fazla ağ çağrısı tetiklememek için olayları gruplayabilir, uzun süren işlem dizilerine zaman aşımı tanımlayabilir ya da gereksiz iş yapmamak için kullanıcı arayüzü olaylarının sıklığını sınırlandırabiliriz.

Bölüm 13’te, “Iterator Trait’i ve next Metodu” başlığında Iterator trait’ini incelerken de bir öğe dizisi görmüştük; ancak iteratörler ile asenkron kanal alıcısı arasında iki temel fark vardır. İlk fark zamandır: iteratörler senkronken kanal alıcısı asenkrondur. İkinci fark ise API’dir. Doğrudan Iterator ile çalışırken onun senkron next metodunu çağırırız. Özel olarak trpl::Receiver akışında ise bunun yerine asenkron bir recv metodu çağırmıştık. Bunların dışında bu API’ler kullanım hissi açısından birbirine çok benzer ve bu benzerlik bir tesadüf değildir. Akış, yinelemenin asenkron bir biçimi gibidir. Bununla birlikte trpl::Receiver özel olarak mesaj almayı beklerken, genel amaçlı akış API’si çok daha kapsamlıdır: tıpkı Iterator’ın yaptığı gibi sıradaki öğeyi sunar, ancak bunu asenkron olarak yapar.

Rust’ta iteratörler ile akışlar arasındaki bu benzerlik, aslında herhangi bir iteratörden bir akış oluşturabileceğimiz anlamına gelir. Bir iteratörde olduğu gibi, bir akışla da next metodunu çağırıp ardından çıktısını await ile bekleyerek çalışabiliriz; henüz derlenmeyecek olan Liste 17-21’de bu kullanım gösterilmektedir.

extern crate trpl; // required for mdbook test

fn main() {
    trpl::block_on(async {
        let degerler = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
        let iter = degerler.iter().map(|n| n * 2);
        let mut akis = trpl::stream_from_iter(iter);

        while let Some(deger) = akis.next().await {
            println!("Değer şuydu: {deger}");
        }
    });
}

Bir sayı dizisiyle başlıyor, bunu bir iteratöre dönüştürüyor ve ardından tüm değerleri iki katına çıkarmak için üzerinde map çağırıyoruz. Daha sonra trpl::stream_from_iter fonksiyonunu kullanarak iteratörü bir akışa dönüştürüyoruz. Ardından bir while let döngüsüyle akıştaki öğeler üzerinde geldikçe dönüyoruz.

Ne yazık ki kodu çalıştırmayı denediğimizde derlenmez ve bunun yerine kullanılabilir bir next metodu olmadığını bildirir:

error[E0599]: no method named `next` found for struct `tokio_stream::iter::Iter` in the current scope
  --> src/main.rs:10:38
   |
10 |         while let Some(deger) = akis.next().await {
   |                                      ^^^^
   |
   = help: items from traits can only be used if the trait is in scope
help: the following traits which provide `next` are implemented but not in scope; perhaps you want to import one of them
   |
1  + use crate::trpl::StreamExt;
   |
1  + use futures_util::stream::stream::StreamExt;
   |
1  + use std::iter::Iterator;
   |
1  + use std::str::pattern::Searcher;
   |
help: there is a method `try_next` with a similar name
   |
10 |         while let Some(deger) = akis.try_next().await {
   |                                      ~~~~~~~~

Bu çıktının da açıkladığı gibi, derleyici hatasının nedeni next metodunu kullanabilmek için doğru trait’in kapsamda olması gerekliliğidir. Şimdiye kadarki tartışmamıza bakarak bu trait’in Stream olmasını makul bir şekilde bekleyebilirsiniz, ancak gereken trait aslında StreamExt’tir. Extension (uzantı/genişletme) kelimesinin kısaltması olan Ext, Rust topluluğunda bir trait’i başka bir trait ile genişletmek için yaygın olarak kullanılan bir desendir.

Stream trait’i, Iterator ve Future trait’lerini etkili bir biçimde birleştiren düşük seviyeli bir arayüz tanımlar. StreamExt ise Stream üzerine, next metodunun yanı sıra Iterator trait’inin sunduklarına benzer diğer yardımcı metotları da içeren daha üst seviyeli bir API kümesi sağlar. Stream ve StreamExt henüz Rust’ın standart kütüphanesinin bir parçası değildir, ancak ekosistemdeki çoğu crate benzer tanımları kullanır.

Derleyici hatasını düzeltmenin yolu, Liste 17-22’deki gibi trpl::StreamExt için bir use deyimi eklemektir.

extern crate trpl; // required for mdbook test

use trpl::StreamExt;

fn main() {
    trpl::block_on(async {
        let degerler = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
        // --snip--
        let iter = degerler.iter().map(|n| n * 2);
        let mut akis = trpl::stream_from_iter(iter);

        while let Some(deger) = akis.next().await {
            println!("Değer şuydu: {deger}");
        }
    });
}

Tüm bu parçalar bir araya geldiğinde kod tam istediğimiz gibi çalışır! Dahası, artık StreamExt kapsamda olduğuna göre tıpkı iteratörlerde olduğu gibi sunduğu tüm yardımcı metotları kullanabiliriz.