Weiterleitung von einem Futures :: Stream zu einem Futures :: Sink

Ich versuche gerade, meinen Kopf um die Tokio & Amp; Futures Primitive und Ökosystem.

Ich begann mit der Arbeit am tk-http websockets Beispiel und wollte mehr Verarbeitung der empfangenen Daten durchführen, als sie zurückzusenden. Ein erster Schritt schien zu sein, den Aufruf von .forward () durch irgendeine Art von Schleife zu ersetzen.

Es scheint mir, dass stream.forward(sink) äquivalent zu stream.fold(sink, |out_, item| { out.send(item).and_then(Sink::flush) }) ist, tue dies jedoch ( commit ) Der Stream wird überhaupt nicht abgefragt. Die gleiche Änderung aus dem Beispiel von Tokio funktioniert jedoch gut ( Beispiel ).

Außerdem scheint es so zu sein, dass die Weiterleitung gut funktioniert, wenn etwas auf der Senke ausgegeben wird ( commit ). Vielleicht wird die Senke aus irgendeinem Grund nicht in der Ereignisschleife registriert, bis ein Sendevorgang stattfindet? Habe ich etwas verpasst? Ist es möglich, dass es ein Fehler in tk-http ist?