// Copyright (c) 2015, the Dart project authors.  Please see the AUTHORS file
// for details. All rights reserved. Use of this source code is governed by a
// BSD-style license that can be found in the LICENSE file.

library async.subscription_stream;

import 'dart:async';

import "delegate/stream_subscription.dart";

/// A [Stream] adapter for a [StreamSubscription].
///
/// This class allows as `StreamSubscription` to be treated as a `Stream`.
///
/// The subscription is paused until the stream is listened to,
/// then it is resumed and the events are passed on to the
/// stream's new subscription.
///
/// This class assumes that is has control over the original subscription.
/// If other code is accessing the subscription, results may be unpredictable.
class SubscriptionStream<T> extends Stream<T> {
  /// The subscription providing the events for this stream.
  StreamSubscription _source;

  /// Create a single-subscription `Stream` from [subscription].
  ///
  /// The `subscription` should not be paused. This class will not resume prior
  /// pauses, so being paused is indistinguishable from not providing any
  /// events.
  ///
  /// If the `subscription` doesn't send any `done` events, neither will this
  /// stream. That may be an issue if `subscription` was made to cancel on
  /// an error.
  SubscriptionStream(StreamSubscription subscription)
       : _source = subscription {
    _source.pause();
    // Clear callbacks to avoid keeping them alive unnecessarily.
    _source.onData(null);
    _source.onError(null);
    _source.onDone(null);
  }

  StreamSubscription<T> listen(void onData(T event),
                               {Function onError,
                                void onDone(),
                                bool cancelOnError}) {
    if (_source == null) {
      throw new StateError("Stream has already been listened to.");
    }
    cancelOnError = (true == cancelOnError);
    var subscription = _source;
    _source = null;
    var result;
    if (cancelOnError) {
      result = new _CancelOnErrorSubscriptionWrapper<T>(subscription);
    } else {
      // Wrap the subscription to ensure correct type parameter.
      result = new DelegatingStreamSubscription<T>(subscription);
    }
    result.onData(onData);
    result.onError(onError);
    result.onDone(onDone);
    subscription.resume();
    return result;
  }
}

/// Subscription wrapper that cancels on error.
///
/// Used by [SubscriptionStream] when forwarding a subscription
/// created with `cancelOnError` as `true` to one with (assumed)
/// `cancelOnError` as `false`. It automatically cancels the
/// source subscription on the first error.
class _CancelOnErrorSubscriptionWrapper<T>
    extends DelegatingStreamSubscription<T> {
  _CancelOnErrorSubscriptionWrapper(StreamSubscription subscription)
      : super(subscription);

  void onError(Function handleError) {
    // Cancel when receiving an error.
    super.onError((error, StackTrace stackTrace) {
      var cancelFuture = super.cancel();
      if (cancelFuture != null) {
        // Wait for the cancel to complete before sending the error event.
        cancelFuture.whenComplete(() {
          if (handleError is ZoneBinaryCallback) {
            handleError(error, stackTrace);
          } else {
            handleError(error);
          }
        });
      } else {
        if (handleError is ZoneBinaryCallback) {
          handleError(error, stackTrace);
        } else {
          handleError(error);
        }
      }
    });
  }
}