Stream's in Dart Part 2

Photo by veeterzy on Unsplash

Im ersten Teil zu Stream's in Dart haben wir erste Schritte in dem Thema unternommen.
Wir erinnern uns, ein Stream, einen Strom also, kann man sich als Fluss vorstellen. Er hat eine Quelle, die ihn fortwährend mit Wasser speist.

Im letzten Post war die Quelle eine Funktion mit der wir Zeichenketten in diesen Fluss eingespeist haben. Der Vergleich mit einem Fluss aus der realen Welt endet hier. Wichtig ist: Wir haben es mit einem fortwährenden Strom von Daten zu tun, die wir manipulieren können.


Das Thema Stream's in Dart ist sehr umfangreich und man läuft Gefahr vor lauter Bäumen den Wald nicht mehr zu sehen. Beginnen wir deshalb einfach:

Stream

Laut Dokumentation ist Stream eine Sequenz an Ereignissen, das kann ein Daten Ereignis sein, oder ein Fehler Ereignis.

Einfach ausgedrückt, jedesmal wenn wir einen Stream auslesen, enthalten wir entweder Daten oder eine Fehlermeldung. Im Sprachgebrauch von Dart "hört" man auf einen Strom, deshalb treffen wir auch auf zum Beispiel stream.listen(...) .
Dart unterscheidet zwei Arten von Streams:

single-subscription stream: hat nur einen "Zuhörer". Der Stream fängt erst dann an Daten zu senden, wenn er einen Zuhörer hat. Der Strom beendet die Lieferung von Daten, wenn sich der Zuhörer vom Strom abmeldet, selbst wenn da noch mehr Daten wären, die er senden könnte.

broadcast stream: Ein Stream der sobald er aktiviert ist Daten sendet, egal ob er Zuhörer hat oder nicht. Das heisst, wenn man so einem Stream zuhört, kann es sein das dieser schon Daten gesendet hat und wenn man sich abmeldet, wird er eventuell weiter Daten senden. Dieser Stream Typ kann beliebig viele Zuhörer haben.

Der single-subscription stream wird oft bei Datei Operationen eingesetzt, der broadcast stream ist für eine allgemeinere Implementation gedacht. Stream's sind dabei im Kontext von asynchronem Code und Futures zu sehen.

Eine andere Definition von Streams, die ebenfalls richtig ist: Streams sind ein Strom an Ergebnissen.
Damit ist gemeint, das die Daten im Stream ja zuerst an einer anderen Stelle im Code generiert wurden. Diese Generierung liefert Ergebnisse, unsere Daten, oder auch Fehler.

Schauen wir uns einmal die Struktur des Stream Komplexes in Dart an:

Allein an diesem Bild sieht man schon, wie Umfangreich, dieses Gebiet ist. Schauen wir uns jetzt noch einmal das leicht abgeänderte Beispiel vom letzten Post an:

import 'dart:async';

List<String> meineStrings = ["Wir versuchen uns", "dem Prinzip von","Streams zu nähern","und es zu verstehen"];

StreamSubscription myabbo;

main() {

  
Duration zeitraum = Duration(seconds: 2);

Stream quelle = Stream.periodic(zeitraum,verarbeiteWerte);

myabbo = quelle.listen(bearbeiteEreignis,onDone: (){
  print('Fertig');
},onError: (err){
  print('Da war ein Fehler im Strom: $err');
},cancelOnError: true);

}

  verarbeiteWerte(int i){
     
    if(i < meineStrings.length){

         return meineStrings[i];
       
    }else{
     
     myabbo.cancel();    
    }
  }

  bearbeiteEreignis(var ereignis){
   
    print(ereignis);

}

Im Code oben benutzen wir:

Stream.periodic
StreamSubscription

Mit Stream.periodic haben wir einen Stream erzeugt, der in der Variable quelle als Stream Objekt gespeichert wird. quelle.listen(...) registriert uns bei diesem Stream und retourniert ein StreamSubscription Objekt, welches wir in myabbo speichern. An einer anderen Stelle im Code oben, beenden wir unser "abbo" mit myabbo.cancel(); .

Stream und async und Future

Im Beispiel Code oben haben wir eine Verzögerung eingebaut (Duration), die wir selbst kontrollieren. In realen Anwendungen werden wir aber oft nicht Kontrolle darüber haben, in welchen zeitlichen Abständen die Daten auf einem Stream eintreffen. Wenn man das bedenkt, drängt es sich fast auf, den entsprechenden Code asynchron zu programmieren. Erzeugen wir einmal einen Stream auf diese Weise:

Future<int> summiereZahlen(Stream<int> stream) async {
  var sum = 0;
  await for (var value in stream) {
    sum += value;
  }
  return sum;
}

Stream<int> zahlenStrom(int to) async* {
  for (int i = 1; i <= to; i++) {
    yield i;
  }
}

main() async {
  var stream = zahlenStrom(10);
  var sum = await summiereZahlen(stream);
  print(sum); // 55
}

Schauen wir uns in Code oben zuerst einmal zahlenStrom an:
Diese Funktion generiert Integer Werte bis zu der Grenze die Ihr als Parameter übergeben wird. Durch das Schlusswort async* zeigen wir an, das die Funktion einen Stream generiert. Die einzelnen Werte geben wir mit yield an die aufrufende Funktion zurück. Die Funktion ist vom Typ Stream<int>

summiereZahlen: eine "normale" async Funktion. Mit await for wird jedes Element des Streams verarbeitet, wobei der Code erst dann mit return sum fortgesetzt wird, wenn stream beendet wurde.

Wir haben jetzt in den zwei Artikeln zwei Varianten gesehen einen Stream zu erzeugen, einmal über die Library und in diesem Post über async, async* und Future<int>.
Man sollte sich nicht entmutigen lassen, das Thema ist komplex und bei falscher Handhabung fehleranfällig.. Aber es lohnt sich.. Bis bald.