How make my own Stream

Question

I have already try to understand the API doc, the articles about them, and this post: How do you create a Stream in Dart

I'm making a simple web app using WebSocket. Actually, it's working well, but I want add a feature (enjoy learn).

This is my class (can be optimized I guess)

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.'));
            });
        });
    }
}

Example of use:

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']);
        });
    });
}

I would replace the then by a listen who will be canceled when the message will completed. By this way, I can add a progress bar, I think...

So my question, my send function can be a stream and keep my concept of one websocket for all ? (yes, if my function is used when a request is in progress, it's sent and if she's finish before the first, I recovered her properly. Thank you ws_id).

Thank you.

Answer 1

I think what you need is a StreamController https://api.dartlang.org/apidocs/channels/stable/dartdoc-viewer/dart-async.StreamController