stream - 在Dart中创建循环流

我尝试编写一个聊天应用程序以与计算机聊天。用户可以编写消息并获得计算机的响应。聊天记录可能如下所示:

user: Hi
computer: Hello
user: What's your name?
computer: Bot
...

我基于循环流的设计灵感来自Cycle.js的思想。我有一个用户消息流，这些用户消息流已转换为计算机消息流，而计算机消息流又是用户消息流的输入:

    |----> user message stream ---->|
    |                               |
transform                       transform
    |                               |
    |<-- computer message stream <--|

该代码已经可以使用:

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

void main() {
  cycle(computer, user);
}

typedef Stream<T> Transform<T>(Stream<T> input);

void cycle(Transform aToB, Transform bToA) {
  var aProxy = new StreamController.broadcast();
  var b = aToB(aProxy.stream);
  var a = bToA(b);
  aProxy.add('start'); // start with user
  aProxy.addStream(a);
}

Stream<String> user(Stream<String> computerMessages) {
  computerMessages = computerMessages.asBroadcastStream();
  computerMessages.listen((message) => print('computer: $message'));
  return computerMessages.map((message) {
    stdout.write('user: ');
    return stdin.readLineSync();
  });
}

Stream<String> computer(Stream<String> userMessages) {
  var messages = <String, String>{
    "Hi": "Hello",
    "What's your name?": "Bot"
  };
  return userMessages.map((m) => messages.containsKey(m) ? messages[m] : 'What?');
}

只有一个问题。您需要一个起始值才能使循环流运行。因此，我将此行放在函数cycle中:

aProxy.add('start'); // start with user

实际上，此逻辑属于我的函数user，因为cycle不应该知道初始值。而且，我不喜欢打印初始值。它仅应触发用户输入流。因此，我更改了cycle和user:

void cycle(Transform aToB, Transform bToA) {
  var aProxy = new StreamController.broadcast();
  var b = aToB(aProxy.stream);
  var a = bToA(b);
  aProxy.addStream(a);
}

Stream<String> user(Stream<String> computerMessages) {
  computerMessages = computerMessages.asBroadcastStream();
  computerMessages.listen((message) => print('computer: $message'));
  var requestInput = new StreamController<String>.broadcast();
  requestInput.add('start'); // start with user
  requestInput.addStream(computerMessages); // continue on computer response
  return requestInput.stream.map((message) {
    stdout.write('user: ');
    return stdin.readLineSync();
  });
}

但是通过此更改，我的应用程序立即终止，且stdout中没有消息。怎么了？

最佳答案

我找到了解决方案。在StreamController中创建一个普通的而不是广播的user:

Stream<String> user(Stream<String> computerMessages) {
  computerMessages = computerMessages.asBroadcastStream();
  computerMessages.listen((message) => print('computer: $message'));
  var requestInput = new StreamController<String>();
  requestInput.add('start'); // start with user
  requestInput.addStream(computerMessages); // continue on computer response
  return requestInput.stream.map((message) {
    stdout.write('user: ');
    return stdin.readLineSync();
  });
}

即使找到了解决方案，但遗憾的是我无法真正解释广播StreamController的问题所在。 null已成功添加requestInput.add(null);，但奇怪的是，既没有完成requestInput.addStream(computerMessages);，也没有将computerMessages事件添加到requestInput中。因此，requestInput关闭，应用程序终止。如果有人可以提供进一步的解释，我将不胜感激。

关于stream - 在Dart中创建循环流，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/31626243/