request_reply_service.dart

import 'dart:async';

import 'package:flutter/foundation.dart';

/// This class allows request-reply communication over messaging.
class RequestReplyService {
  StreamSubscription _subscription;

  /// release resources
  void dispose() {
    _subscription?.cancel();
  }

  /// send request and return response object as [Future]
  Future<T> execute<T>({
    @required Future<void> Function() request,
    @required Stream<T> Function() response,
  }) async {
    final completer = Completer<T>();

    // close the last subscription which may be open after
    // fast subsequent calls of this method
    _subscription?.cancel();

    _subscription = response()
        .handleError((error) => completer.completeError(error))
        .listen((T data) {
      if (!completer.isCompleted) {
        completer.complete(data);
      }
    });    
    request();

    return completer.future;
  }
}