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

WFRxDocStoreChangesFeedImpTest.lua

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

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

NOTICE: All information contained herein is, and remain 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 may be covered by U.S. and Foreign
Patents, patents in process, and are protected by trade secret or copyright
law.  Dissemination of this information or reproduction of this material is
strictly forbidden unless prior written permission is obtained from
Adobe Systems Incorporated.

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

local Rx = require 'Rx'

	-- WF.core.rxlua
local WFRxScheduler = require 'WFRxScheduler'

	-- WF.core.self_test
local WFRxTestCase = require 'WFRxTestCase'

	-- WF.core.utils
local WFCloseSupport = require 'WFCloseSupport'

	--
local WFRxDocStoreChangesFeedImp = require 'WFRxDocStoreChangesFeedImp'

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

local function setUpDocStore( rxTest )

	local docStore = require "WFRxDocStore"( ':memory:' )

	rxTest:finally( function()

		if docStore and not WFCloseSupport.hasBeenOrWillBeClosed( docStore ) then

			local waitForThis = docStore:performIntegrityCheck()

			local function closeWhenDone()
				docStore:close()
				docStore = nil
			end

			return waitForThis:finally( closeWhenDone )

		end

	end )

	return docStore

end

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

local function setDocValueAndValidateRevId( rxTest, docStore, docValue )

	local revId

	rxTest:expectSingleValue(

		docStore:setDocValue( docValue )
			:map( function( rev )
				revId = rev
				return not not rev:match( '^[0-9a-f]+$' ), #rev, rev
			end ),

		true, 32, ':ignore:' )

	return revId

end

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

local function rxTestBogusFuncSpecs( rxTest )

	local docStore = setUpDocStore( rxTest )

	local feed

	feed = rxTest:observe( docStore:scannedFilteredMappedChangesFeedSince( 
											"wut", "wut", "what", -1 ) )

	feed:expectError( ':ignore:' )

	feed = rxTest:observe( docStore:scannedFilteredMappedChangesFeedSince( 
								"WFRxDocStoreChangesFeedImpTest", "wut", 
								"what", -1 ) )

	feed:expectError( ':ignore:' )

	feed = rxTest:observe( docStore:scannedFilteredMappedChangesFeed( 
											"wut", "wut", "what" ) )

	feed:expectError( ':ignore:' )

	feed = rxTest:observe( docStore:scannedFilteredMappedChangesFeed( 
								"WFRxDocStoreChangesFeedImpTest", "wut",
								"what" ) )

	feed:expectError( ':ignore:' )

end

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

local function docValueFieldMatches( wantedInfo, docValue )

	local fieldName, wantedValue = unpack( wantedInfo )

	if type( docValue ) ~= "table" or
				docValue[ fieldName ] ~= wantedValue then

		return wantedInfo

	end

	return wantedInfo, docValue

end

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

local function rxTestFilteredChangesFeed( rxTest )

	local docStore = setUpDocStore( rxTest )

	local evenFeed = rxTest:observe( 
						docStore:scannedFilteredMappedChangesFeed( 
							"WFRxDocStoreChangesFeedImpTest", 
							"docValueFieldMatches", 
							{ "isEven", "yup" } ) )

	local oddFeed = rxTest:observe( 
						docStore:scannedFilteredMappedChangesFeed( 
							"WFRxDocStoreChangesFeedImpTest", 
							"docValueFieldMatches", 
							{ "isEven", "nope" } ) )

	local biggishFeed = rxTest:observe( 
						docStore:scannedFilteredMappedChangesFeed( 
							"WFRxDocStoreChangesFeedImpTest", 
							"docValueFieldMatches", 
							{ "isSmall", "nope" } ) )

	local revIds = {}

	for i = 1, 10 do

		revIds[ i ] = setDocValueAndValidateRevId( rxTest, docStore,
							{
								_id = tostring( i ),
								isEven = i % 2 == 0 and "yup" or "nope",
								isSmall = i < 10 and "yup" or "nope",
							} )

	end

	for i = 1, 10 do

		local expectedFeed = i % 2 == 0 and evenFeed or oddFeed

		local expectedIsEven = i % 2 == 0 and "yup" or "nope"

		expectedFeed:expectData {
				_id = tostring( i ),
				_localId = i,
				_changeCounter = i,
				_revId = revIds[ i ],
				isEven = expectedIsEven,
				isSmall = i < 10 and "yup" or "nope",
			}

		if i == 10 then

			biggishFeed:expectData {
					_id = tostring( i ),
					_localId = i,
					_changeCounter = i,
					_revId = revIds[ i ],
					isEven = expectedIsEven,
					isSmall = "nope",
				}

		end

	end

	local i = 11

	revIds[ i ] = setDocValueAndValidateRevId( rxTest, docStore,
						{
							_id = tostring( i ),
							isEven = "nope",
							isSmall = "nope",
						} )

	oddFeed:expectData {
			_id = tostring( i ),
			_localId = i,
			_changeCounter = i,
			_revId = revIds[ i ],
			isEven = "nope",
			isSmall = "nope",
		}

	biggishFeed:expectData {
			_id = tostring( i ),
			_localId = i,
			_changeCounter = i,
			_revId = revIds[ i ],
			isEven = "nope",
			isSmall = "nope",
		}

end

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

local function stringifyScannedDataValue( data, docValue )

	return string.format( "%s[ %d ] = %d%s", 
							data.wut, data.seen, docValue._localId, docValue.isFresh or "" )

end

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

local function scanEvenDocValues( data, docValue )

	if docValue._localId % 2 == 1 then

		return data

	end

	data.seen = ( data.seen or 0 ) + 1

	return data, stringifyScannedDataValue( data, docValue )

end

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

local function scanOddDocValues( data, docValue )

	if docValue._localId % 2 == 0 then

		return data

	end

	data.seen = ( data.seen or 0 ) + 1

	return data, stringifyScannedDataValue( data, docValue )

end

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

local function rxTestScannedDistinctChangesFeed( rxTest )

	local docStore = setUpDocStore( rxTest )

	local evenFeed = rxTest:observe( 
						docStore
							:scannedFilteredMappedChangesFeed( 
								"WFRxDocStoreChangesFeedImpTest", 
								"scanEvenDocValues", 
								{ wut = "isEven" } ) )

	local oddFeed = rxTest:observe( 
						docStore
							:scannedFilteredMappedChangesFeed( 
								"WFRxDocStoreChangesFeedImpTest", 
								"scanOddDocValues", 
								{ wut = "isOdd" } ) )

	for i = 1, 10 do

		setDocValueAndValidateRevId( rxTest, docStore,
							{
								_id = tostring( i ),
								isEven = i % 2 == 0 and "yup" or "nope",
							} )

	end


	local expectedEvenResults = {
			"isEven[ 1 ] = 2",
			"isEven[ 2 ] = 4",
			"isEven[ 3 ] = 6",
			"isEven[ 4 ] = 8",
			"isEven[ 5 ] = 10" }

	local expectedOddResults = {
			"isOdd[ 1 ] = 1",
			"isOdd[ 2 ] = 3",
			"isOdd[ 3 ] = 5",
			"isOdd[ 4 ] = 7", 
			"isOdd[ 5 ] = 9" }

	for i = 1, 5 do

		rxTest:logf( "Checking for result %d", i )

		evenFeed:expectData( expectedEvenResults[ i ] )

		oddFeed:expectData( expectedOddResults[ i ] )

	end

end

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

local function rxTestScannedDistinctChangesFeed_NoChanges( rxTest )

	local timeLimit = 0.5

	-- expecting 4 timeouts so test has to run at least that long
	rxTest:setTimeLimit( 2 + 4 * timeLimit )

	local docStore = setUpDocStore( rxTest )


	local evenFeed = rxTest:observe(
						docStore
							:scannedFilteredMappedChangesFeed(
								"WFRxDocStoreChangesFeedImpTest",
								"scanEvenDocValues",
								{ wut = "isEven" } ) )

	local oddFeed = rxTest:observe(
						docStore
							:scannedFilteredMappedChangesFeed(
								"WFRxDocStoreChangesFeedImpTest",
								"scanOddDocValues",
								{ wut = "isOdd" } ) )

	local function makeEvenOddObservers()

		local evenFeedWithTimeout = rxTest:observe(
					docStore
						:scannedFilteredMappedChangesFeed(
							"WFRxDocStoreChangesFeedImpTest",
							"scanEvenDocValues",
							{ wut = "isEven" } )
						:timeout( timeLimit, WFRxScheduler.sharedScheduler() ) )

		local oddFeedWithTimeout = rxTest:observe(
					docStore
						:scannedFilteredMappedChangesFeed(
							"WFRxDocStoreChangesFeedImpTest",
							"scanOddDocValues",
							{ wut = "isOdd" } )
						:timeout( timeLimit, WFRxScheduler.sharedScheduler() ) )

		return evenFeedWithTimeout, oddFeedWithTimeout

	end


	-- Make sure nothing is produced with an empty DocStore
	-- and the error is the expected timeout.

	do
		local evenFeedWithTimeout, oddFeedWithTimeout = makeEvenOddObservers( rxTest, docStore )

		oddFeedWithTimeout:expectError( "Timeout" )
		evenFeedWithTimeout:expectError( "Timeout" )

	end


	-- Make sure no even changes are produced when an odd value is added.

	do

		local evenFeedWithTimeout, oddFeedWithTimeout = makeEvenOddObservers( rxTest, docStore )

		-- See if it supplies nothing when only an odd value is added.
		local i = 1
		setDocValueAndValidateRevId( rxTest, docStore,
			{
				_id = tostring( i ),
				isEven = i % 2 == 0 and "yup" or "nope",
			} )

		oddFeedWithTimeout:expectData( "isOdd[ 1 ] = 1" )
		evenFeedWithTimeout:expectError( "Timeout" )

	end


	-- Make sure no odd changes are produced when an even value is added.

	do

		local evenFeedWithTimeout, oddFeedWithTimeout = makeEvenOddObservers( rxTest, docStore )

		-- See if it supplies nothing when only an odd value is added.
		local i = 2
		setDocValueAndValidateRevId( rxTest, docStore,
			{
				_id = tostring( i ),
				isEven = i % 2 == 0 and "yup" or "nope",
			} )

		evenFeedWithTimeout:expectData( "isEven[ 1 ] = 2" )
		oddFeedWithTimeout:expectError( "Timeout" )

	end


	-- Make sure nothing is produced with a populated DocStore
	-- and the error is the expected timeout.

	do

		local evenFeedWithTimeout, oddFeedWithTimeout = makeEvenOddObservers( rxTest, docStore )

		evenFeedWithTimeout:expectError( "Timeout" )
		oddFeedWithTimeout:expectError( "Timeout" )

	end


	-- feeds started at beginning saw the values
	oddFeed:expectData( "isOdd[ 1 ] = 1" )
	evenFeed:expectData( "isEven[ 1 ] = 2" )

end

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

local function rxTestScannedDistinctChangesFeedSince( rxTest )

	local docStore = setUpDocStore( rxTest )

	for i = 1, 10 do

		setDocValueAndValidateRevId( rxTest, docStore,
							{
								_id = tostring( i ),
								isEven = i % 2 == 0 and "yup" or "nope",
							} )

	end

	local evenFeed = rxTest:observe( 
						docStore
							:scannedFilteredMappedChangesFeedSince( 
								"WFRxDocStoreChangesFeedImpTest", 
								"scanEvenDocValues", 
								{ wut = "isEven" },
								5 ) )

	local oddFeed = rxTest:observe( 
						docStore
							:scannedFilteredMappedChangesFeedSince( 
								"WFRxDocStoreChangesFeedImpTest", 
								"scanOddDocValues", 
								{ wut = "isOdd" },
								5 ) )


	local expectedEvenResults = {
			"isEven[ 1 ] = 6",
			"isEven[ 2 ] = 8",
			"isEven[ 3 ] = 10" }

	local expectedOddResults = {
			"isOdd[ 1 ] = 5",
			"isOdd[ 2 ] = 7", 
			"isOdd[ 3 ] = 9" }

	for i = 1, 3 do

		rxTest:logf( "Checking for result %d", i )

		evenFeed:expectData( expectedEvenResults[ i ] )

		oddFeed:expectData( expectedOddResults[ i ] )

	end

end

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

local function rxTestElideStaleDocsChangesFeedSince( rxTest )

	local docStore = setUpDocStore( rxTest )

	-- Set up scannedFilteredMappedChangesFeedSince. Whatever document
	-- revisions are processed will end up in the interal buffer.
	
	local oddFeed = rxTest:observe( 
						docStore
							:scannedFilteredMappedChangesFeedSince( 
								"WFRxDocStoreChangesFeedImpTest", 
								"scanOddDocValues", 
								{ wut = "isOdd" },
								0 ) )
								:map( function( docValue )
									return docValue._id
								end )

	-- Add a test doc and listen for it on the feed. We do this
	-- to make sure that `WFRxDocStoreChangesFeedImp` is fully
	-- inited before we move on to the rest of the test.
	
	setDocValueAndValidateRevId( rxTest, docStore,
			{
				_id = 99,
				isEven = "nope",
				isFresh = ":FRESH"
			} )

	oddFeed:expectData( '99' )

	for i = 1, 10 do

		-- Create a document, and a new revision.
		
		setDocValueAndValidateRevId( rxTest, docStore,
				{
					_id = tostring( i ),
					isEven = i % 2 == 0 and "yup" or "nope",
					isFresh = ":STALE"
				} )

		rxTest:waitFor(

			docStore:applyOverlay( {
				_id = tostring( i ),
				isFresh = ":FRESH"
			} )
			
		)
			
	end

	-- Setting up a new scannedFilteredMappedChangesFeedSince should fire
	-- with both the revisions in the internal buffer, as well as all the
	-- new revisions.
	
	local evenFeedAgain = rxTest:observe( 
						docStore
							:scannedFilteredMappedChangesFeedSince( 
								"WFRxDocStoreChangesFeedImpTest", 
								"scanEvenDocValues", 
								{ wut = "isEven" },
								0 ) )

	local expectedEvenResults = {
		"isEven[ 1 ] = 2:STALE",
		"isEven[ 2 ] = 2:FRESH",
		"isEven[ 3 ] = 4:STALE",
		"isEven[ 4 ] = 4:FRESH",
		"isEven[ 5 ] = 6:STALE",
		"isEven[ 6 ] = 6:FRESH",
		"isEven[ 7 ] = 8:STALE",
		"isEven[ 8 ] = 8:FRESH",
		"isEven[ 9 ] = 10:STALE",
		"isEven[ 10 ] = 10:FRESH",
	}

	for i = 1, 10 do

		rxTest:logf( "Checking for result %d", i )

		evenFeedAgain:expectData( expectedEvenResults[ i ] )

	end

	-- Using scannedFilteredMappedChangesFeedSinceWithoutStaleRevisions, the stale
	-- document revisions should be elided from the feed.

	local elidedEvenFeed = rxTest:observe( 
						docStore
							:scannedFilteredMappedChangesFeedSinceWithoutStaleRevisions( 
								"WFRxDocStoreChangesFeedImpTest", 
								"scanEvenDocValues", 
								{ wut = "isEven" },
								0 ) )

	expectedEvenResults = {
		"isEven[ 1 ] = 2:FRESH",
		"isEven[ 2 ] = 4:FRESH",
		"isEven[ 3 ] = 6:FRESH",
		"isEven[ 4 ] = 8:FRESH",
		"isEven[ 5 ] = 10:FRESH",
	}
	
	for i = 1, 5 do

		rxTest:logf( "Checking for result %d", i )

		elidedEvenFeed:expectData( expectedEvenResults[ i ] )

	end

	oddFeed:dispose()

end

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

local function rxTestElideStaleDocsChangesFeedSince_CatchUp( rxTest )

	local docStore = setUpDocStore( rxTest )

	for i = 1, 10 do

		-- Create a document, and a new revision.
		
		setDocValueAndValidateRevId( rxTest, docStore,
				{
					_id = tostring( i ),
					isEven = i % 2 == 0 and "yup" or "nope",
					isFresh = ":STALE"
				} )

		rxTest:waitFor(

			docStore:applyOverlay( {
				_id = tostring( i ),
				isFresh = ":FRESH"
			} )
			
		)
			
	end

	-- Using scannedFilteredMappedChangesFeedSinceWithoutStaleRevisions, the stale
	-- document revisions should be elided from the feed. Since there 
	-- were no previous subscribers to WFRxDocStoreChangesFeedImp, the
	-- internal catch-up mechanism should be used.

	local elidedEvenFeed = rxTest:observe( 
						docStore
							:scannedFilteredMappedChangesFeedSinceWithoutStaleRevisions( 
								"WFRxDocStoreChangesFeedImpTest", 
								"scanEvenDocValues", 
								{ wut = "isEven" },
								0 ) )

	local expectedEvenResults = {
		"isEven[ 1 ] = 2:FRESH",
		"isEven[ 2 ] = 4:FRESH",
		"isEven[ 3 ] = 6:FRESH",
		"isEven[ 4 ] = 8:FRESH",
		"isEven[ 5 ] = 10:FRESH",
	}
	
	for i = 1, 5 do

		rxTest:logf( "Checking for result %d", i )

		elidedEvenFeed:expectData( expectedEvenResults[ i ] )

	end

end

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

local function scanAllDocValues( data, docValue )

	return data, docValue

end

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

local function rxTestElideStaleDocsChangesFeedSince_Conflict( rxTest )

	local docStore = setUpDocStore( rxTest )

	-- Set up scannedFilteredMappedChangesFeedSince. Whatever document
	-- revisions are processed will end up in the interal buffer.
	
	local oddFeed = rxTest:observe( 
						docStore
							:scannedFilteredMappedChangesFeedSince( 
								"WFRxDocStoreChangesFeedImpTest", 
								"scanOddDocValues", 
								{ wut = "isOdd" },
								0 ) )
								:map( function( docValue )
									return docValue._id
								end )

	-- Add a test doc and listen for it on the feed. We do this
	-- to make sure that `WFRxDocStoreChangesFeedImp` is fully
	-- inited before we move on to the rest of the test.
	
	setDocValueAndValidateRevId( rxTest, docStore,
			{
				_id = 99,
				isEven = "nope",
				isFresh = ":FRESH"
			} )

	oddFeed:expectData( '99' )

	local revIds = {}

	for i = 1, 2 do

		-- Create a few documents
		
		revIds[ i ] = setDocValueAndValidateRevId( rxTest, docStore,
				{
					_id = tostring( i ),
					isEven = i % 2 == 0 and "yup" or "nope",
					isFresh = i % 2 == 0 and "STALE" or "FRESH"
				} )
			
	end

	-- Create a conflict on a document.

	local setRev2a = {
		_id = "2",
		isEven = "yup",
		_revId = revIds[ 2 ],
		isFresh = "CONFLICT1"
	}
	
	local secondRevId = rxTest:getFirstValue( docStore:setDocValue( setRev2a ) )

	-- Add a conflicting revision derived from first.

	local conflictingRevId = require 'KSIdGenerator'.uuidInLowercaseHex()

	local setRev2b = {
		_id = "2",
		isEven = "yup",
		_revIds = { revIds[ 2 ], conflictingRevId },
		isFresh = "CONFLICT2"
	}
	
	local secondRevIdB = rxTest:getFirstValue( docStore:setDocValue( setRev2b ) )
	
	-- Verify that asset document has two head revisions.

	rxTest:expectSingleValue(

		docStore:currentRevisionsForDoc( "2" ),

		{
			{
				_changeCounter = 3,
				_conflict = true,
				_id = "2",
				_localId = 2,
				_revId = secondRevId,
				isEven = "yup",
				isFresh = "CONFLICT1"
			},
			{
				_changeCounter = 4,
				_conflict = true,
				_id = "2",
				_localId = 2,
				_revId = secondRevIdB,
				isEven = "yup",
				isFresh = "CONFLICT2"
			},
		} )
	
	-- Using scannedFilteredMappedChangesFeedSinceWithoutStaleRevisions, the stale
	-- document revisions should be elided from the feed.

	local elidedFeed = rxTest:observe( 
						docStore
							:scannedFilteredMappedChangesFeedSinceWithoutStaleRevisions( 
								"WFRxDocStoreChangesFeedImpTest", 
								"scanAllDocValues", 
								{ wut = "isEven" },
								0 ) )

	local expectedResults = { 
		{
			_changeCounter = 1,
			_id = "1",
			_localId = 1,
			_revId = revIds[ 1 ],
			isEven = "nope",
			isFresh = "FRESH",
		},
		{
			_changeCounter = 4,
			_conflict = true,
			_id = "2",
			_localId = 2,
			_revId = secondRevId,
			isEven = "yup",
			isFresh = "CONFLICT1",
		}
	}
	
	elidedFeed:expectData( expectedResults[ 1 ] )
	elidedFeed:expectData( expectedResults[ 2 ] )
	
	-- This should match the output of changedDocs

	expectedResults.continueFrom = 5
	
	rxTest:expectSingleValue(
		docStore:changedDocs { fromChangeCounter = 0 }, 
		expectedResults )
	
	oddFeed:dispose()

end

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

local function rxTestScannedDistinctChangesFeedSince_OneBufferedChange( rxTest )

	local docStore = setUpDocStore( rxTest )

	local oddFeed1 = rxTest:observe( 
						docStore
							:scannedFilteredMappedChangesFeedSince( 
								"WFRxDocStoreChangesFeedImpTest", 
								"scanOddDocValues", 
								{ wut = "isOdd" },
								1 ) )

	setDocValueAndValidateRevId( rxTest, docStore,
			{
				_id = tostring( 1 ),
				isEven = "nope",
			} )

	local expectedOddResults = {"isOdd[ 1 ] = 1" }

	rxTest:logf( "Checking for result %d", 1 )

	oddFeed1:expectData( expectedOddResults[ 1 ] )

	local oddFeed2 = rxTest:observe( 
						docStore
							:scannedFilteredMappedChangesFeedSince( 
								"WFRxDocStoreChangesFeedImpTest", 
								"scanOddDocValues", 
								{ wut = "isOdd" },
								1 ) )

	rxTest:logf( "Checking for result %d", 1 )

	oddFeed2:expectData( expectedOddResults[ 1 ] )

end

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

local function watchForWutChanges( wantedInfo, docValue )

	local localAssetId = docValue._localId

	if wantedInfo[ localAssetId ] ~= docValue.wut then
		
		wantedInfo[ localAssetId ] = docValue.wut
	
		return wantedInfo, localAssetId, docValue.wut

	end

	return wantedInfo

end

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

local function rxTestEmitOnlyOnChange( rxTest )

	local docStore = setUpDocStore( rxTest )

	local wutFeed = rxTest:observe( 
						docStore:scannedFilteredMappedChangesFeed( 
							"WFRxDocStoreChangesFeedImpTest", 
							"watchForWutChanges", {} ) )

	for i = 1, 10 do

		setDocValueAndValidateRevId( rxTest, docStore,
							{
								_id = tostring( i ),
								wut = "wot",
							} )

	end

	for i = 1, 10 do

		wutFeed:expectData( i, "wot" )

	end

	for i = 1, 10 do

		rxTest:expectSingleValue(

			docStore:applyOverlay( 
							{
								_id = tostring( i ),
								wut = i > 5 and "wat" or "wot",
							} ),
			':ignore:' )

	end

	for i = 1, 10 do

		if i > 5 then

			wutFeed:expectData( i, "wat" )

		end

	end

end

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

local function makeBufferElement( i )
	return {
		_changeCounter = i,
	}
end

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

local function rxTestInternalBuffer( rxTest )

	local TestAPI = WFRxDocStoreChangesFeedImp._expose_test_API()

	for n = 5, 6 do

		local buffer = TestAPI.createChangesFeedBuffer( n )

		TestAPI.appendDocValueToBuffer( buffer, makeBufferElement( 1 ) )

		rxTest:expectEqual( { makeBufferElement( 1 ) },
							TestAPI.getChangedDocsFromBufferOffset(
								buffer,
								TestAPI.getBufferOffset( buffer, 0 ) ) )

		local expectedChangedDocs = {}

		-- wrapped buffer
		for i = 1, 2 * n do

			TestAPI.appendDocValueToBuffer( buffer, makeBufferElement( i ) )

			if i > n then

				expectedChangedDocs[ #expectedChangedDocs + 1 ] = 
					makeBufferElement( i )

			end

		end

		for i = 1, n do
			rxTest:expectEqual( { unpack( expectedChangedDocs, i, n ) },
								TestAPI.getChangedDocsFromBufferOffset(
									buffer,
									TestAPI.getBufferOffset( buffer, i - 1 ) ) )
		end

		expectedChangedDocs = {}

		for i = 1, n  do

			buffer = TestAPI.createChangesFeedBuffer( n )

			for j = 1, n + i - 1 do

				TestAPI.appendDocValueToBuffer( buffer, makeBufferElement( 2 * j ) )

			end

			for j = i, n + i - 1 do
				
				rxTest:expectEqual( 
					j <= n and j or j - n,
					TestAPI.getBufferOffsetForChangeCounter( buffer, 2 * j ) )

			end

		end

		for i = 1, n  do

			buffer = TestAPI.createChangesFeedBuffer( n )

			for j = 1, n + i - 1 do

				TestAPI.appendDocValueToBuffer( buffer, makeBufferElement( 2 * j ) )

			end

			local _, compareResult = TestAPI.getBufferOffsetForChangeCounter( buffer, 2 * i - 1 )

			rxTest:expectEqual( -1, compareResult )

			_, compareResult = TestAPI.getBufferOffsetForChangeCounter( buffer, 2 * ( n + i ) - 1 )

			rxTest:expectEqual( 1, compareResult )

			for j = i + 1, n + i - 1 do
				
				rxTest:expectEqual( 
					j <= n and j or j - n,
					TestAPI.getBufferOffsetForChangeCounter( buffer, 2 * j - 1 ) )

			end

		end

	end

	local n = 5

	local buffer = TestAPI.createChangesFeedBuffer( n )

	local beforeChangeCounter, buffereredChangedDocs, futureChangeCounter =
				TestAPI.maybeGetBufferedChanges( buffer, 1 )

	rxTest:expectEqual( nil, beforeChangeCounter )
	rxTest:expectEqual( nil, buffereredChangedDocs )
	rxTest:expectEqual( nil, futureChangeCounter )

	local expectedChangedDocs = {}

	local m = 2

	for i = 1, n do

		local element = makeBufferElement( i + m )

		expectedChangedDocs[ #expectedChangedDocs + 1 ] = element

		TestAPI.appendDocValueToBuffer( buffer, element )

	end

	beforeChangeCounter, buffereredChangedDocs, futureChangeCounter =
				TestAPI.maybeGetBufferedChanges( buffer,  m )

	rxTest:expectEqual( m, beforeChangeCounter )

	rxTest:expectEqual( expectedChangedDocs, buffereredChangedDocs )

	rxTest:expectEqual( n + m + 1, futureChangeCounter )

	beforeChangeCounter, buffereredChangedDocs, futureChangeCounter =
				TestAPI.maybeGetBufferedChanges( buffer,  n + m + 1 )

	rxTest:expectEqual( nil, beforeChangeCounter )

	rxTest:expectEqual( nil, buffereredChangedDocs )

	rxTest:expectEqual( n + m + 1, futureChangeCounter )

	beforeChangeCounter, buffereredChangedDocs, futureChangeCounter =
				TestAPI.maybeGetBufferedChanges( buffer,  n + m + 2 )

	rxTest:expectEqual( nil, beforeChangeCounter )

	rxTest:expectEqual( nil, buffereredChangedDocs )

	rxTest:expectEqual( n + m + 2, futureChangeCounter )

	for i = 1, n do

		beforeChangeCounter, buffereredChangedDocs, futureChangeCounter =
					TestAPI.maybeGetBufferedChanges( buffer,  i + m )

		rxTest:expectEqual( nil, beforeChangeCounter )

		rxTest:expectEqual( { unpack( expectedChangedDocs, i, n ) },
							buffereredChangedDocs  )

		rxTest:expectEqual( n + m + 1, futureChangeCounter )

	end

end

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

local function makeMockDocStoreImp( initialChangeCounter )

	return {

		_changesFeed = Rx.Subject(),

		_raw_changesFeed = function( self )

			return self._changesFeed

		end,

		_changeCounter = Rx.BehaviorSubject( 
							tonumber( initialChangeCounter ) or 1 ),

		changeCounter = function( self )

			return self._changeCounter
			
		end,

		_changedDocsCalls = {},

		changedDocs = function( self, continueFromArg )
				
			local chgDocsSubj = Rx.Subject()

			self._changedDocsCalls[ #self._changedDocsCalls + 1 ] = 
				{ continueFromArg.fromChangeCounter, chgDocsSubj }

			return chgDocsSubj

		end, 

		request = function( self, api, arg )

			return self[ api ]( self, arg )

		end,

		close = function() end,

	}

end

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

local function makeJustReturn( dsImp )

	return function()
	
		return dsImp

	end

end

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

local function makeNRecords( n )

	local records = {}

	local changeCounterToRecordLookup = {}

	local currChangeCounter = 1

	for _ = 1, n do

		records[ #records + 1 ] = { _changeCounter = currChangeCounter }

		changeCounterToRecordLookup[ currChangeCounter ] = #records

		currChangeCounter = currChangeCounter + math.random( 3 )

	end

	return records, changeCounterToRecordLookup

end

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

local function getFeedResults( feed )

	local results = {}

	results.sub = feed:subscribe(
						function( dv )

							results[ #results + 1 ] = dv

						end )

	return results

end

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

local function getChangedDocsDocs( changeDocs, start, finish )

	local changeCounter = changeDocs[ finish + 1 ]._changeCounter

	return { continueFrom = changeCounter, 
				unpack( changeDocs, start, finish ) }, changeCounter, finish - start

end

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

local function rxTestChangesFeedStream( rxTest )

	rxTest:setTimeLimit( 360 )

	local TestAPI = WFRxDocStoreChangesFeedImp._expose_test_API()

	local prevBufferSize = TestAPI.injectAlternateBufferSize( 3 )

	local prevRawDocFn = TestAPI.injectAlternateProcessChangedDocsRawDocRow( function( ... )
			return ...
		end )

	local masterChangeDocs = makeNRecords( 1000 )

	local initialDocOffset = 5

	local currentDocOffset = initialDocOffset

	local initialChangeCounter = 
			masterChangeDocs[ initialDocOffset ]._changeCounter

	local mockDSImp = makeMockDocStoreImp( initialChangeCounter )

	local prevCreator = TestAPI.injectAlternateDocStoreImpCreator( 
							makeJustReturn( mockDSImp ) )

	rxTest:finally( 
			function()
		
				TestAPI.injectAlternateBufferSize( prevBufferSize )

				TestAPI.injectAlternateDocStoreImpCreator( prevCreator )
				
				TestAPI.injectAlternateProcessChangedDocsRawDocRow( prevRawDocFn )

			end )

	local wfRxDocStoreChangesFeedImp = WFRxDocStoreChangesFeedImp( 'wut' )

	local futureFeed = 
			getFeedResults( 
				TestAPI.changesFeedSinceStream( 
					wfRxDocStoreChangesFeedImp, 
					masterChangeDocs[ currentDocOffset + 1 ]._changeCounter ) )

	rxTest:expectEqual( 0, #futureFeed )

	rxTest:expectEqual(
			0,
			#mockDSImp._changedDocsCalls )

	local needsChangedDocsFromOffset1Feed = 
			getFeedResults( TestAPI.changesFeedSinceStream( 
									wfRxDocStoreChangesFeedImp, 1 ) )

	rxTest:expectEqual( 0, #needsChangedDocsFromOffset1Feed )

	rxTest:expectEqual(
			1,
			#mockDSImp._changedDocsCalls )

	rxTest:expectEqual(
			1,
			mockDSImp._changedDocsCalls[ 1 ][ 1 ] )

	-- send 3 docs
	local firstThreeDocs = getChangedDocsDocs( masterChangeDocs, 1, 3 )

	mockDSImp._changedDocsCalls[ 1 ][ 2 ]:onNext(
				getChangedDocsDocs( masterChangeDocs, 1, 3 ) )

	rxTest:expectEqual( 0, #futureFeed )

	rxTest:expectEqual( 3, #needsChangedDocsFromOffset1Feed )

	rxTest:expectEqual(
			2,
			#mockDSImp._changedDocsCalls )

	rxTest:expectEqual(
			firstThreeDocs.continueFrom,
			mockDSImp._changedDocsCalls[ 2 ][ 1 ] )

	--send rest of docs
	mockDSImp._changedDocsCalls[ 2 ][ 2 ]:onNext(
				getChangedDocsDocs( masterChangeDocs, 4, 5 ) )

	rxTest:expectEqual( 0, #futureFeed )

	rxTest:expectEqual(
			2,
			#mockDSImp._changedDocsCalls )

	rxTest:expectEqual(
			{ unpack( masterChangeDocs, 1, 5 ) },
			{ unpack( needsChangedDocsFromOffset1Feed, 1, 5 ) } )

	local numNewRecords = 4

	for i = currentDocOffset + 1, currentDocOffset + numNewRecords - 1 do

		mockDSImp._changesFeed:onNext( { masterChangeDocs[ i ] } )

	end
	
	currentDocOffset = currentDocOffset + numNewRecords - 1

	rxTest:expectEqual(
			2,
			#mockDSImp._changedDocsCalls )

	rxTest:expectEqual(
			{ unpack( masterChangeDocs, initialDocOffset + 1, currentDocOffset ) },
			{ unpack( futureFeed, 1, currentDocOffset - initialDocOffset + 1 ) } )

	rxTest:expectEqual(
			{ unpack( masterChangeDocs, 1, currentDocOffset ) },
			{ unpack( needsChangedDocsFromOffset1Feed, 1, currentDocOffset ) } )

	local bufferedFeed = 
			getFeedResults( 
				TestAPI.changesFeedSinceStream( 
					wfRxDocStoreChangesFeedImp, 
					masterChangeDocs[ initialDocOffset + 1 ]._changeCounter ) )

	rxTest:expectEqual(
			2,
			#mockDSImp._changedDocsCalls )

	rxTest:expectEqual(
			{ unpack( masterChangeDocs, initialDocOffset + 1, currentDocOffset ) },
			{ unpack( bufferedFeed, 1, currentDocOffset - initialDocOffset + 1 ) } )

	local needsChangedDocsFromOffset2Feed = 
			getFeedResults( TestAPI.changesFeedSinceStream( 
									wfRxDocStoreChangesFeedImp, 2 ) )

	rxTest:expectEqual( 0, #needsChangedDocsFromOffset2Feed )

	rxTest:expectEqual(
			3,
			#mockDSImp._changedDocsCalls )

	--send catchup docs 
	mockDSImp._changedDocsCalls[ 3 ][ 2 ]:onNext(
				getChangedDocsDocs( masterChangeDocs, 2, currentDocOffset ) )

	rxTest:expectEqual(
			{ unpack( masterChangeDocs, 2, currentDocOffset ) },
			{ unpack( needsChangedDocsFromOffset2Feed, 1, currentDocOffset - 1 ) } )

	local needsChangedDocsFromOffset3Feed = 
			getFeedResults( TestAPI.changesFeedSinceStream( 
									wfRxDocStoreChangesFeedImp, 3 ) )

	rxTest:expectEqual( 0, #needsChangedDocsFromOffset3Feed )

	rxTest:expectEqual(
			4,
			#mockDSImp._changedDocsCalls )

	numNewRecords = 14

	for i = currentDocOffset + 1, currentDocOffset + numNewRecords - 1 do

		mockDSImp._changesFeed:onNext( { masterChangeDocs[ i ] } )

	end
	
	currentDocOffset = currentDocOffset + numNewRecords - 1

	rxTest:expectEqual(
			{ unpack( masterChangeDocs, initialDocOffset + 1, currentDocOffset ) },
			{ unpack( futureFeed ) } )

	rxTest:expectEqual(
			{ unpack( masterChangeDocs, 1, currentDocOffset ) },
			{ unpack( needsChangedDocsFromOffset1Feed ) } )

	rxTest:expectEqual(
			{ unpack( masterChangeDocs, 2, currentDocOffset ) },
			{ unpack( needsChangedDocsFromOffset2Feed ) } )

	rxTest:expectEqual(
			{ unpack( masterChangeDocs, initialDocOffset + 1, currentDocOffset ) },
			{ unpack( bufferedFeed ) } )

	--send catchup docs 
	mockDSImp._changedDocsCalls[ 4 ][ 2 ]:onNext(
				getChangedDocsDocs( masterChangeDocs, 3, initialDocOffset ) )

	rxTest:expectEqual(
			{ unpack( masterChangeDocs, 3, currentDocOffset ) },
			{ unpack( needsChangedDocsFromOffset3Feed ) } )

	numNewRecords = 144

	for i = currentDocOffset + 1, currentDocOffset + numNewRecords - 1 do

		mockDSImp._changesFeed:onNext( { masterChangeDocs[ i ] } )

	end
	
	currentDocOffset = currentDocOffset + numNewRecords - 1

	rxTest:expectEqual(
			{ unpack( masterChangeDocs, initialDocOffset + 1, currentDocOffset ) },
			{ unpack( futureFeed ) } )

	rxTest:expectEqual(
			{ unpack( masterChangeDocs, 1, currentDocOffset ) },
			{ unpack( needsChangedDocsFromOffset1Feed ) } )

	rxTest:expectEqual(
			{ unpack( masterChangeDocs, 2, currentDocOffset ) },
			{ unpack( needsChangedDocsFromOffset2Feed ) } )

	rxTest:expectEqual(
			{ unpack( masterChangeDocs, initialDocOffset + 1, currentDocOffset ) },
			{ unpack( bufferedFeed ) } )

	rxTest:expectEqual(
			{ unpack( masterChangeDocs, 3, currentDocOffset ) },
			{ unpack( needsChangedDocsFromOffset3Feed ) } )

end

--==============================================================================

return WFRxTestCase.wrapSuite( 'WFRxDocStoreChangesFeedImpTest', {

	rxTestBogusFuncSpecs = rxTestBogusFuncSpecs,

	rxTestFilteredChangesFeed = rxTestFilteredChangesFeed,

	rxTestScannedDistinctChangesFeed = rxTestScannedDistinctChangesFeed,

	rxTestScannedDistinctChangesFeed_NoChanges = rxTestScannedDistinctChangesFeed_NoChanges,

	rxTestScannedDistinctChangesFeedSince = rxTestScannedDistinctChangesFeedSince,

	rxTestScannedDistinctChangesFeedSince_OneBufferedChange = rxTestScannedDistinctChangesFeedSince_OneBufferedChange,

	rxTestEmitOnlyOnChange = rxTestEmitOnlyOnChange,

	rxTestInternalBuffer = rxTestInternalBuffer,

	rxTestChangesFeedStream = rxTestChangesFeedStream,

	rxTestElideStaleDocsChangesFeedSince = rxTestElideStaleDocsChangesFeedSince,

	rxTestElideStaleDocsChangesFeedSince_CatchUp = rxTestElideStaleDocsChangesFeedSince_CatchUp,

	rxTestElideStaleDocsChangesFeedSince_Conflict = rxTestElideStaleDocsChangesFeedSince_Conflict,

	-- exported functions imported by this test
	docValueFieldMatches = docValueFieldMatches,

	scanEvenDocValues = scanEvenDocValues,

	scanOddDocValues = scanOddDocValues,

	scanAllDocValues = scanAllDocValues,

	watchForWutChanges = watchForWutChanges,

} )
