--[[----------------------------------------------------------------------------

WFCatImportManifestProcessor.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2017 Adobe Systems Incorporated
All Rights Reserved.

NOTICE: All information contained herein is, and remains the property of
Adobe Systems Incorporated and its suppliers, if any. The intellectual and
technical concepts contained herein are proprietary to Adobe Systems
Incorporated and its suppliers and are protected by all applicable
intellectual property laws, including trade secret and copyright laws.
Dissemination of this information or reproduction of this material is
strictly forbidden unless prior written permission is obtained from
Adobe Systems Incorporated.

----------------------------------------------------------------------------]]--

	--Rx
local Rx = require 'Rx'
local Observable = Rx.Observable

	-- WF.oz_client.managed_catalog
local WFCatImportManifestThreadDB = require 'WFCatImportManifestThreadDB'
local WFCatImportManifestStages = require 'WFCatImportManifestStages'
local WFCatImportManifestIdCache = require 'WFCatImportManifestIdCache'
local WFCatImportManifestMessages = require 'WFCatImportManifestMessages'
local WFCatImportManifestConstants = require 'WFCatImportManifestConstants'

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

local kProcessorChainPause = "processor_chain_pause"

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

local WFCatImportManifestProcessor = {}

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

local function combineStageInfo( stages, catalog, profileCatalog, wfCatSynchronizer, 
									manifestDatabase, manifestStatusDatabase, idCache, messageHandler, contextInfo )

	local obs = Observable.just( true )
	
	for _, stage in ipairs( stages ) do
		messageHandler:registerStage( stage.name, stage.progressWeight )
		
		local processorInstance = stage.processor()
		
		-- Give a chance for the processor to populate the cache
		if processorInstance:hasMethod( 'setup' ) then
			obs = obs
				:flatMap( function()
					
					local setupObservable = processorInstance:setup( catalog, manifestDatabase, manifestStatusDatabase, idCache, messageHandler )
					
					return setupObservable
						:flatMap( function( success )
							if success then
								return Observable.just( true )
							else
								return Observable.throw( kProcessorChainPause )
							end
						end )
					
				end )
		end
		
		-- Let the processor do the bulk of it's processing
		obs = obs
			:flatMap( function()
				
				local processObservable = processorInstance:process( catalog, profileCatalog, 
					wfCatSynchronizer, manifestDatabase, manifestStatusDatabase, idCache, messageHandler, contextInfo )
				
				return processObservable
					:flatMap( function( success )
					
						if success then
							return Observable.just( true )
						else
							return Observable.throw( kProcessorChainPause )
						end
					end )
				
			end )
	end
	
	return obs
end

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

function WFCatImportManifestProcessor.importManifest( path, fileName, catalog, profileCatalog, wfCatSynchronizer, contextInfo )

	local manifestDatabase = WFCatImportManifestThreadDB( "WFCatImportManifest.lua", path, fileName )
	local manifestStatusDatabase = WFCatImportManifestThreadDB( "WFCatImportManifestStatus.lua", path, fileName )
	
	local idCache = WFCatImportManifestIdCache()
	
	local messageHandler = WFCatImportManifestMessages()
	
	local stages = WFCatImportManifestStages.getStages()
	
	local importStagesObservable = combineStageInfo( stages, catalog, profileCatalog, 
										wfCatSynchronizer, manifestDatabase, manifestStatusDatabase, 
										idCache, messageHandler, contextInfo )
		:tapOnCompleted( function()
			messageHandler:sendStatus( { status = WFCatImportManifestConstants.importStatus.completed, } )
		end )
		:catch( function( err )
			
			-- If we are throwing the err for breaking the chain, 
			-- then eat it up silently and close.
			if err == kProcessorChainPause then
				return Observable.empty()
			end
			
			-- This is a valid err. Rethrow it.
			return Observable.throw( err )
			
		end )
		:finally( function()
			manifestDatabase:close()
			manifestStatusDatabase:close()
			messageHandler:shutdown()
		end )
	
	return messageHandler:messages()
		:merge( importStagesObservable:ignoreElements() )
		
end

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

return WFCatImportManifestProcessor
