local Rx = require 'Rx'

local AbstractObserver = Rx.AbstractObserver
local CompositeDisposable = Rx.CompositeDisposable
local Notification = Rx.Notification
local ObservableBase = Rx.ObservableBase

local notificationCreateOnNext = Notification.createOnNext
local notificationCreateOnError = Notification.createOnError
local notificationCreateOnCompleted = Notification.createOnCompleted

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

local MaterializeObserver = Rx.Class 'Rx.internals.MaterializeObserver' {

    AbstractObserver,
        -- superclass

    _observer = false,
        -- (Observer)

    __init = function(self, observer)

        self._observer = observer

    end,

    next = function(self, ...)

        self._observer:onNext(notificationCreateOnNext(...))

    end,

    error = function(self, ...)

        self._observer:onNext(notificationCreateOnError(...))
        self._observer:onCompleted()

    end,

    completed = function(self)

        self._observer:onNext(notificationCreateOnCompleted())
        self._observer:onCompleted()

    end,

}

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

local MaterializeObservable = Rx.Class 'Rx.internals.MaterializeObservable' {

    ObservableBase,
        -- superclass

    _source = false,
        -- (Observable)

    __init = function(self, source)

        self._source = source

    end,

    subscribeCore = function(self, observer)

        local materializeObserver = MaterializeObserver(observer)

        return CompositeDisposable(self._source:subscribe(materializeObserver),
            materializeObserver)

    end,

}

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

return function(self)

    return MaterializeObservable(self)

end
