local Rx = require 'Rx'

local CompositeDisposable = Rx.CompositeDisposable
local SingleAssignmentDisposable = Rx.SingleAssignmentDisposable
local AnonymousObservable = Rx.AnonymousObservable

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

return function(self, ...)

    if select('#', ...) > 0 then
        error("observable:flatMergeAll() accepts no parameters. (Maybe you meant :flatMerge(n)?)")
    end

    return AnonymousObservable(function(observer)

        local group = CompositeDisposable()
        local isStopped = false
        local m = SingleAssignmentDisposable()

        group:add(m)

        m:setDisposable(self:subscribe(
            function(innerSource)

                local innerSubscription = SingleAssignmentDisposable()
                group:add(innerSubscription)

                innerSubscription:setDisposable(innerSource:subscribe(
                    function(...)
                        return observer:onNext(...)
                    end,
                    function(...)
                        return observer:onError(...)
                    end,
                    function()
                        group:remove(innerSubscription)
                        if isStopped and group:length() == 1 then
                            observer:onCompleted()
                        end
                    end))

            end,
            function(...)
                return observer:onError(...)
            end,
            function()
                isStopped = true
                if group:length() == 1 then
                    observer:onCompleted()
                end
            end))

        return group

    end)

end
