async Future StreamSubscription Error

Question

Не могли бы вы объяснить, что не так в следующем коде. Я делаю два вызова функции fInputData. Первое работает нормально, второе приводит к ошибке: «необработанное исключение» «Плохое состояние: у потока уже есть подписчик»

Мне нужно написать тестовую консольную программу, которая вводит несколько параметров.

import "dart:async" as async; 
import "dart:io"; 

void main() { 
    fInputData ("Enter Nr of Iterations : ") 
    .then((String sResult){ 
    int iIters; 
    try { 
     iIters = int.parse(sResult); 
     if (iIters < 0) throw new Exception("Invalid"); 
    } catch (oError) { 
     print ("Invalid entry"); 
     exit(1); 
    } 

    print ("In Main : Iterations selected = ${iIters}"); 

    fInputData("Continue Processing? (Y/N) : ") // this call bombs 
    .then((String sInput){ 
     if (sInput != "y" && sInput != "Y") 
     exit(1); 

     fProcessData(iIters); 

     print ("Main Completed"); 
    }); 
    }); 
} 

async.Future<String> fInputData(String sPrompt) { 
    async.Completer<String> oCompleter = new async.Completer(); 

    stdout.write(sPrompt); 
    async.Stream<String> oStream = stdin.transform(new StringDecoder()); 
    async.StreamSubscription oSub; 
    oSub = oStream.listen((String sInput) { 
    oCompleter.complete(sInput); 
    oSub.cancel(); 
    }); 
    return oCompleter.future; 
} 

void fProcessData(int iIters) { 
    print ("In fProcessData"); 
    print ("iIters = ${iIters}"); 

    for (int iPos = 1; iPos <= iIters; iPos++) { 
    if (iPos%100 == 0) print ("Processed = ${iPos}"); 
    } 
    print ("In fProcessData - completed ${iIters}"); 
} 

Answer 1

Некоторые background reading:

Потоки поставляется в двух вариантах: один или несколько (также известный как вещания) абонента. По умолчанию наш поток представляет собой поток с одним абонентом . Это означает, что , если вы попытаетесь прослушивать поток более один раз, вы получите исключение и с помощью любой функции обратного вызова функции или будущие свойства учитываются как прослушивание.

Вы можете преобразовать однопользовательский поток в широковещательный поток с помощью метода asBroadcastStream().

Таким образом, у вас есть два варианта: либо повторно использовать один объект подписки. т. е. вызывать прослушивание один раз и сохранить объект подписки живым.

Или используйте широковещательный поток - обратите внимание, что существует ряд различий между широковещательными потоками и потоками с одним абонентом, вам нужно будет прочитать об этом и убедиться, что они соответствуют вашему прецеденту.

Вот пример повторного использования абонентом задать несколько вопросов:

import 'dart:async'; 
import 'dart:io'; 

main() { 
    var console = new Console(); 
    var loop; 
    loop =() => ask(console).then((_) => loop()); 
    loop(); 
} 

Future ask(Console console) { 
    print('1 + 1 = ...'); 
    return console.readLine().then((line) { 
     print(line.trim() == '2' ? 'Yup!' : 'Nope :('); 
    }); 
} 

class Console { 
    StreamSubscription<String> _subs; 

    Console() { 
     var input = stdin 
      .transform(new StringDecoder()) 
      .transform(new LineTransformer()); 

     _subs = input.listen(null); 
    } 

    Future<String> readLine() { 
     var completer = new Completer<String>(); 
     _subs.onData(completer.complete); 
     return completer.future; 
    } 
}