如何制作我自己的 Stream

Question

我已经尝试了解 API 文档、有关它们的文章以及这篇文章：How do you create a Stream in Dart

我正在使用 WebSocket 制作一个简单的 Web 应用程序。实际上，它运行良好，但我想添加一个功能（享受学习）。

这是我的课（我猜可以优化）

library Ask;

import 'dart:html';
import 'dart:async';
import 'dart:convert';

class Ask {
    final String addr;
    String _protocol;
    String _port;
    WebSocket _ws;
    bool openned;
    Map<int, Completer> _completer_list = {};
    int _counter = 0;
    static final Map<String, Ask> _cache = <String, Ask>{};

    factory Ask(String addr) {
        if (_cache.containsKey(addr)) {
            return _cache[addr];
        } else {
            final ask_server = new Ask._internal(addr);
            _cache[addr] = ask_server;
            return ask_server;
        }
    }

    Ask._internal(this.addr);

    Future<bool> open() {
        if (openned)
            return true;
        _completer_list[0] = new Completer();
        if (window.location.protocol == 'http:') {
            _port = ':8080/ws';
            _protocol = 'ws://';
        } else {
            _port = ':8443/ws';
            _protocol = 'wss://';
        }
        _ws = new WebSocket(_protocol + addr + _port);
        _ws.onOpen.listen((e) {
            _get_data();
            _get_close();
            openned = true;
            _completer_list[0].complete(true);
        });
        return _completer_list[0].future;
    }

    Future<String> send(Map data) {
        bool check = false;
        int id;

        _completer_list.forEach((k, v) {
            if (v.isCompleted) {
                id = data['ws_id'] = k;
                _completer_list[k] = new Completer();
                _ws.send(JSON.encode(data));
                check = true;
            }
        });
        if (!check) {
            _counter++;
            id = data['ws_id'] = _counter;
            _completer_list[id] = new Completer();
            _ws.send(JSON.encode(data));
        }
        return _completer_list[id].future;
    }

    void _get_data() {
        _ws.onMessage.listen((MessageEvent data) {
            var response = JSON.decode(data.data);
            _completer_list[response['ws_id']].complete(response);
        });
    }

    void _get_close() {
        _ws.onClose.listen((_) {
            print('Server have been lost. Try to reconnect in 3 seconds.');
            new Timer(new Duration(seconds: 3), () {
                _ws = new WebSocket(_protocol + addr + _port);
                _get_data();
                _get_close();
                _ws.onOpen.listen((e) => print('Server is alive again.'));
            });
        });
    }
}

使用示例：

void showIndex() {
    Element main = querySelector('main');
    Ask connect = new Ask('127.0.0.1');
    Map request = {};

    request['index'] = true;
    connect.open().then((_) {
        connect.send(request).then((data) {
            main.setInnerHtml(data['response']);
        });
    });
}

我会将then 替换为listen，当消息完成时，该listen 将被取消。这样一来，我就可以添加进度条了，我觉得……

所以我的问题是，我的发送函数可以是一个流并保持我的一个 websocket 的概念吗？ （是的，如果在请求进行时使用了我的函数，它会被发送，如果她在第一次之前完成，我会正确恢复她。谢谢 ws_id）。

谢谢。

Answer 1

我认为你需要的是一个 StreamController https://api.dartlang.org/apidocs/channels/stable/dartdoc-viewer/dart-async.StreamController