--------------------------------------------------------------------------------

timeoutWith
  :: (TimeoutOptions a -> TimeoutOptions a)
     -> Observable Async a
     -> Observable Async a
timeoutWith modFn source =
    Observable $ \observer -> do
      sourceDisposable <- newSingleAssignmentDisposable
      timerDisposable  <- newBooleanDisposable
      subscription <-
        main sourceDisposable timerDisposable observer

      setDisposable sourceDisposable subscription

      return $ toDisposable sourceDisposable `mappend`
               toDisposable timerDisposable
  where
    defOpts = TimeoutOptions {
        _timeoutInterval = seconds 0
      , _timeoutStartAfterFirst = False
      , _completeOnTimeout = False
      , _resetTimeoutWhen = const True
      , _timeoutScheduler = newThread
      }
    opts = modFn defOpts
    scheduler = _timeoutScheduler opts
    interval  = _timeoutInterval opts
    shouldResetTimeout = _resetTimeoutWhen opts

    main sourceDisposable timerDisposable observer = do
        resetTimeout
        subscribe source onNext_ onError_ onCompleted_
      where
        onTimeout =
          if _completeOnTimeout opts
            then onCompleted observer
            else onError observer $ toException TimeoutError
        resetTimeout = do
          timer <- scheduleTimed scheduler interval $ do
            onTimeout
            void $ dispose sourceDisposable
          -- This will automatically dispose the previous timer
          setDisposable timerDisposable timer

        onNext_ v = do
          onNext observer v
          when (shouldResetTimeout v) resetTimeout
        onError_ = onError observer
        onCompleted_ = onCompleted observer
timeoutWith
  :: (TimeoutOptions a -> TimeoutOptions a)
     -> Observable Async a
     -> Observable Async a
timeoutWith modFn source =
    Observable $ \observer -> do
      sourceDisposable <- newSingleAssignmentDisposable
      timerDisposable  <- newBooleanDisposable
      subscription <-
        main sourceDisposable timerDisposable observer

      setDisposable sourceDisposable subscription

      return $ toDisposable sourceDisposable `mappend`
               toDisposable timerDisposable
  where
    defOpts = TimeoutOptions {
        _timeoutInterval = seconds 0
      , _timeoutStartAfterFirst = False
      , _completeOnTimeout = False
      , _resetTimeoutWhen = const True
      , _timeoutScheduler = newThread
      }
    opts = modFn defOpts
    scheduler = _timeoutScheduler opts
    interval  = _timeoutInterval opts
    shouldResetTimeout = _resetTimeoutWhen opts

    main sourceDisposable timerDisposable observer = do
        resetTimeout
        subscribe source onNext_ onError_ onCompleted_
      where
        onTimeout =
          if _completeOnTimeout opts
            then onCompleted observer
            else onError observer $ toException TimeoutError
        resetTimeout = do
          timer <- scheduleTimed scheduler interval $ do
            onTimeout
            void $ dispose sourceDisposable
          -- This will automatically dispose the previous timer
          setDisposable timerDisposable timer

        onNext_ v = do
          onNext observer v
          when (shouldResetTimeout v) resetTimeout
        onError_ = onError observer
        onCompleted_ = onCompleted observer
timeoutWith modFn source =
    Observable $ \observer -> do
      sourceDisposable <- newSingleAssignmentDisposable
      timerDisposable  <- newBooleanDisposable
      subscription <-
        main sourceDisposable timerDisposable observer

      setDisposable sourceDisposable subscription

      return $ toDisposable sourceDisposable `mappend`
               toDisposable timerDisposable
  where
    defOpts = TimeoutOptions {
        _timeoutInterval = seconds 0
      , _timeoutStartAfterFirst = False
      , _completeOnTimeout = False
      , _resetTimeoutWhen = const True
      , _timeoutScheduler = newThread
      }
    opts = modFn defOpts
    scheduler = _timeoutScheduler opts
    interval  = _timeoutInterval opts
    shouldResetTimeout = _resetTimeoutWhen opts

    main sourceDisposable timerDisposable observer = do
        resetTimeout
        subscribe source onNext_ onError_ onCompleted_
      where
        onTimeout =
          if _completeOnTimeout opts
            then onCompleted observer
            else onError observer $ toException TimeoutError
        resetTimeout = do
          timer <- scheduleTimed scheduler interval $ do
            onTimeout
            void $ dispose sourceDisposable
          -- This will automatically dispose the previous timer
          setDisposable timerDisposable timer

        onNext_ v = do
          onNext observer v
          when (shouldResetTimeout v) resetTimeout
        onError_ = onError observer
        onCompleted_ = onCompleted observer