flutter-sport/lib/provider/message_model.dart

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

import 'package:flutter/cupertino.dart';
import 'package:shared_preferences/shared_preferences.dart';
import 'package:sport/bean/message.dart';
import 'package:sport/db/message_db.dart';
import 'package:sport/services/api/inject_api.dart';

class MessageModel extends ChangeNotifier with InjectApi {
//  Message _message;
  int _curId;
  List<MessageInstance> _messages;

  final StreamController<List<MessageInstance>> _queryController =
      StreamController.broadcast();

  Stream<List<MessageInstance>> get queryStream => _queryController.stream;

  ///事件订阅对象
  StreamSubscription dataSubscription;
  StreamSubscription periodicSubscription;

  init() {
    periodicSubscription =
        Stream.periodic(Duration(seconds: 10)).listen((event) async {
      Message data;

      SharedPreferences prefs = await SharedPreferences.getInstance();
      String token = prefs.getString("token");
      if(token == null || token.isEmpty == true)
        return;

      if (_curId != null) {
        data = (await api.getMessageForPoll(curId: _curId)).data;
      } else {
        data = (await api.getMessageForPoll()).data;
      }
      // curId 这里一定是接口返回的 本地存 sqlite的 curId 全部都是 0
      _curId = data.curId;
      _messages = data.messages;

      // 没有新消息还处理流程吗？
      if(!(data?.messages?.isNotEmpty == true))
        return

//      var list = await MessageDB().findAll(); // 清空数字
//      print("[list]:$list--------------------------");

      // var userList = await MessageDB().findAllUser();
      // print("[list]:$userList--------------------------");
//      MessageDB().deleteTable(); //
      // 这里本来是做了去重操作的 现在不用了 ...
//      if (_messages.length > 0) {
//        var list = await MessageDB().findLatest();
//        print(list);
//        if (list.length != 0) {
//          int lastCurId = list[0]["curId"];
//          print('[lastCurId]:$lastCurId-------------------------');
//          // 可能会重复插入...;
//          if (lastCurId != _curId) {
//            add(data.messages, _curId);
//          }
//        } else {
////          add(data.messages, _curId);
//        }
//      }
      add(data.messages, _curId);
    });
  }

  add(List<MessageInstance> messages, int curId) async {
    List<MessageItem> _items = [];
    // 插进去是 1 未读... 0 已读...
    for (MessageInstance _instance in _messages) {
      _items.add(new MessageItem(
          message: _instance,
          curId: curId,
          status: 1,
          userId: _instance.fromUser.id));
      var list = await MessageDB().findHasUserId(_instance.fromUser.id);
//      print("[user:]$list-------------------------------");
//
//      printWell("list", list);

      if (list.length == 0) {
        await MessageDB().insertUser(
            new UserTableInfo(userId: _instance.fromUser.id, isTop: 0));
      }
    }
    await MessageDB().insertAll(_items);

    // 这里就是 广播?
    get(messages);
//    print("[items]:$_items----------------------------------------------");
  }

  get(List<MessageInstance> list) {
    _queryController.add(list);
  }

  close() {
    ///关闭
    dataSubscription.cancel();
//    _queryController.close();
  }
}

// Generic Interface for all BLoCs
abstract class BlocBase {
  void dispose();
}

// Generic BLoC provider
class BlocProvider<T extends BlocBase> extends StatefulWidget {
  BlocProvider({
    Key key,
    @required this.child,
    @required this.bloc,
  }) : super(key: key);

  final T bloc;
  final Widget child;

  @override
  _BlocProviderState<T> createState() => _BlocProviderState<T>();

  static T of<T extends BlocBase>(BuildContext context) {
    final type = _typeOf<BlocProvider<T>>();
    BlocProvider<T> provider = context.ancestorWidgetOfExactType(type);
    return provider.bloc;
  }

  static Type _typeOf<T>() => T;
}

class _BlocProviderState<T> extends State<BlocProvider<BlocBase>> {
  @override
  void dispose() {
    widget.bloc.dispose();
    super.dispose();
  }

  @override
  Widget build(BuildContext context) {
    return widget.child;
  }
}

class MessageBloc implements BlocBase {
  Message _message;

  //
  // Stream to handle the counter
  //
  StreamController<Message> _counterController = StreamController<Message>();

  // 流的输入...
  StreamSink<Message> get _addOperate => _counterController.sink;
  Stream<Message> get outCounter => _counterController.stream;

  void dispose() {
    _counterController.close();
  }

  // 这是操作?
  void addOperate(data) {
    _addOperate.add(_message);
  }
}