concatMany transforms an IEnumerable to an IObservable. concatMany selects each Enumerable value into an Observable, waits for the IObservable to complete, then selects the next Enumerable value.

concatMany.as

package trxcllnt.vr
{
	import raix.interactive.IEnumerable;
	import raix.interactive.IEnumerator;
	import raix.reactive.CompositeCancelable;
	import raix.reactive.ICancelable;
	import raix.reactive.IObservable;
	import raix.reactive.IObserver;
	import raix.reactive.Observable;
	import raix.reactive.scheduling.Scheduler;

	/**
	 * @author ptaylor
	 */
	public function concatMany(enumerable:IEnumerable, selector:Function):IObservable {
		return Observable.createWithCancelable(function(observer:IObserver):ICancelable {
			
			const iterator:IEnumerator = enumerable.getEnumerator();
			const subscriptions:CompositeCancelable = new CompositeCancelable();
			
			var schedule:Function = function():void {
				subscriptions.add(Scheduler.scheduleRecursive(Scheduler.defaultScheduler, function(reschedule:Function):void {
					
					schedule = reschedule;
					
					const item:Object = iterator.current;
					const obs:IObservable = selector(item);
					
					const completed:Function = function():void {
						subscriptions.remove(subscription);
						recurse();
					};
					
					const subscription:ICancelable = obs.subscribe(observer.onNext, completed, observer.onError);
					
					subscriptions.add(subscription);
				}));
			};
			
			const recurse:Function = ifElse(
				iterator.moveNext, 
				function():void { schedule(); },
				observer.onCompleted
			);
			
			recurse();
			
			return subscriptions;
		});
	}
}

Main.as

package
{
	import asx.fn.K;
	import asx.fn._;
	import asx.fn.args;
	import asx.fn.getProperty;
	import asx.fn.noop;
	import asx.fn.partial;
	import asx.fn.sequence;
	import asx.number.mul;
	
	import flash.display.DisplayObject;
	import flash.display.Graphics;
	import flash.display.Sprite;
	import flash.events.*;
	import flash.geom.Point;
	import flash.geom.Rectangle;
	
	import org.tinytlf.events.mouse.*;
	import org.tinytlf.handlers.printComplete;
	import org.tinytlf.handlers.printError;
	import org.tinytlf.types.Virtualizer;
	
	import raix.interactive.Enumerable;
	import raix.interactive.IEnumerable;
	import raix.interactive.toEnumerable;
	import raix.reactive.IObservable;
	import raix.reactive.ISubject;
	import raix.reactive.Observable;
	import raix.reactive.Subject;
	
	import trxcllnt.vr.virtualize;
	
	[SWF(width = "500", height = "600")]
	public class Main extends Sprite
	{
		public function Main()
		{
			// Init a container sprite.
			const container:Sprite = addChild(new Sprite()) as Sprite;
			
			// Init two subjects, one to dispatch Arrays of items, the other
			// to dispatch viewport Rectangle updates.
			const items:ISubject = new Subject();
			const viewports:ISubject = new Subject();
			
			const dispatches:IObservable = items.
				combineLatest(viewports, args).
				mappend(K(new Virtualizer()));
			
			const updates:IObservable = virtualize(
				{},
				dispatches,
				selectVisible,
				partial(paragraph, 500),
				update,
				partial(layout, _, _, _, _, container)
			);
			
			updates.subscribe(noop, printComplete('updates'), printError('updates'));
			viewports.subscribe(function(rect:Rectangle):void {
				container.scrollRect = rect;
				
				const g:Graphics = container.graphics;
				g.clear();
				g.beginFill(0x00, 0.1);
				g.drawRect(rect.x, rect.y, rect.width, rect.height);
				g.endFill();
				
			}, printComplete('viewports'), printError('viewports'));
			
			const rect:Rectangle = new Rectangle(0, 0, 500, 600);
			
			doDrag(viewports, rect);
			
			viewports.onNext(rect);
			items.onNext([children]);
		}
		
		private function doDrag(viewports:ISubject, rect:Rectangle):void {
			const self:DisplayObject = this;
			const downObs:IObservable = down(this);
			const upObs:IObservable = up(this);
			
			const dragObs:IObservable = downObs.mapMany(function(d:MouseEvent):IObservable {
				
				const moveObs:IObservable = org.tinytlf.events.mouse.move(self);
				
				return moveObs.startWith(d).
					scan(function(o:Object, m:MouseEvent):Object {
						return {
							x: m.stageX,
							y: m.stageY,
							dx: m.stageX - o.x,
							dy: m.stageY - o.y
						};
					}, {x: d.stageX, y: d.stageY}, true).
					map(function(o:Object):Point {
						return new Point(o.dx, o.dy);
					}).
					takeUntil(upObs);
			});
			
			const dragY:IObservable = dragObs.map(getProperty('y'));
			const wheelY:IObservable = Observable.fromEvent(this, MouseEvent.MOUSE_WHEEL).map(getProperty('delta'));
			
			dragY.merge(wheelY).
				map(partial(mul, -1)).
				peek(partial(rect.offset, 0)).
				map(K(rect)).
				subscribe(function(r:Rectangle):void {
					viewports.onNext(r);
				});
		}
	}
}
import asx.array.first;
import asx.array.forEach;
import asx.array.last;
import asx.array.map;
import asx.fn.I;
import asx.fn.K;
import asx.fn._;
import asx.fn.callProperty;
import asx.fn.partial;
import asx.fn.sequence;
import asx.string.Lorem;

import flash.display.DisplayObject;
import flash.display.DisplayObjectContainer;
import flash.display.Sprite;
import flash.geom.Rectangle;
import flash.text.engine.ElementFormat;
import flash.text.engine.TextBlock;
import flash.text.engine.TextElement;
import flash.text.engine.TextLine;
import flash.text.engine.TextLineCreationResult;
import flash.text.engine.TextLineValidity;

import org.tinytlf.types.Virtualizer;

import raix.interactive.Enumerable;
import raix.interactive.IEnumerable;
import raix.interactive.toEnumerable;
import raix.reactive.IObservable;
import raix.reactive.Observable;
import raix.reactive.scheduling.Scheduler;

internal const children:Array = Lorem.paragraphs(1000, 4).map(function(p:String, i:int, ...args):Array {
	p = i + ': ' + p;
	return [p.substring(0, p.indexOf(' ', p.indexOf(' ') + 1)), p];
});

internal function selectVisible(list:Array,
								viewport:Rectangle,
								cache:Virtualizer):IEnumerable {
	
	const cached:Array = cache.slice(viewport.y, viewport.y + viewport.height);
	const end:Object = last(cached);
	const start:int = end == null ? 0 : cache.getIndex(end);
	
	return toEnumerable(cached).skipLast(1).
		map(cache.getIndex).
		map(function(i:int):Object { return list[i]; }).
		concat(Enumerable.generate(
			start,
			function(i:int):Boolean { return i < list.length;},
			function(i:int):int { return i + 1; },
			function(i:int):Object { return list[i]; }
		)).
		takeWhile(partial(haltIteration, viewport, cache));
}

internal function haltIteration(viewport:Rectangle, cache:Virtualizer, element:Object):Boolean {
	
	// If there's still room in the viewport, render the next element.
	if(cache.size <= viewport.bottom) return true;
	
	if(cache.getIndex(element) != -1) {
		// If the element is cached and it intersects with the
		// viewport, render it.
		
		const bounds:Rectangle = new Rectangle(
			viewport.x,
			cache.getStart(element),
			viewport.width,
			cache.getSize(element)
		);
		
		return viewport.intersects(bounds);
	}
	
	return false;
}

internal function update(viewport:Rectangle, cache:Virtualizer, unit:Object, child:DisplayObject):void {
	const size:Number = child.height;
	const index:int = cache.getIndex(unit);
	
	if(index == -1) cache.add(unit, size);
	else cache.setSizeAt(index, size);
}

internal const sprites:Object = {};

internal function paragraph(width:Number, unit:Array):IObservable {
	
	const name:String = first(unit) as String;
	const text:String = last(unit) as String;
	
	if(name in sprites) {
		return Observable.value([unit, sprites[name]]);
	}
	
	const sprite:Sprite = new Sprite();
	sprites[name] = sprite;
	sprite.name = name;
	sprite.removeChildren();
	
	return Observable.value(new TextBlock(new TextElement(text, new ElementFormat()))).
		switchMany(sequence(
			partial(lines, width),
			callProperty('peek', sprite.addChild),
			callProperty('toArray')
		)).
		switchMany(partial(iterateLayout, _, positionLines)).
		map(K([unit, sprite]));
};

internal function layout(viewport:Rectangle, cache:Virtualizer, unit:Object, children:Array, container:DisplayObjectContainer):IObservable {
	
	container.removeChildren();
	forEach(map(children, last), container.addChild);
	
	return iterateLayout(children, partial(positionParagraphs, cache)).map(K([unit, container]));
};

internal function iterateLayout(children:Array, reduction:Function):IObservable {
	return Observable.fromArray(children).
		reduce(reduction, 0, true).
		onErrorResumeNext(Observable.value(0));
};

//internal function positionParagraphs(cache:Virtualizer, y:Number, child:DisplayObject):Number {
internal function positionParagraphs(cache:Virtualizer, y:Number, info:Array):Number {
	const unit:Object = info[0];
	const child:DisplayObject = info[1];
	
	return child.y = cache.getStart(unit);
};

internal function positionLines(y:Number, line:TextLine):Number {
	line.y = y + line.ascent;
	return y + line.textHeight;
};

internal function lines(width:Number, block:TextBlock):IObservable {
	
	var breakAnother:Boolean = false;
	
	const predicate:Function = function(line:TextLine):Boolean {
		return breakAnother;
	};
	
	const iterate:Function = function(line:TextLine):TextLine {
		
		// gimme me a textline.
		line = createTextLine(block, line, width);
		
		// break another while the block is still invalid.
		breakAnother = isBlockInvalid(block);
		
		return line;
	};
	
	// If the block is invalid, it's possible there's some valid lines.
	// Don't re-render all the lines if we don't have to.
	if(isBlockInvalid(block)) {
		
		// Get valid lines to start from.
		const validLines:Array = getValidLines(block);
		
		// Check in old lines.
		// TextLines.checkIn.apply(null, getInvalidLines(block));
		
		// Start from a line, but can be null.
		const initial:TextLine = getLineBeforeFirstInvalidLine(block);
		
		// Concat the valid and new lines together.
		return Observable.concat([
			Observable.fromArray(validLines),
			Observable.generate(initial || iterate(null), predicate, iterate, I)
		]);
	}
	
	breakAnother = true;
	
	// Render all the lines.
	return Observable.generate(iterate(null), predicate, iterate, I, Scheduler.greenThread);
};

internal function createTextLine(block:TextBlock, line:TextLine, width:Number):TextLine {
	return block.createTextLine(line, width, 0.0, true);
};

internal function isBlockInvalid(block:TextBlock):Boolean {
	return block.firstLine == null ||
		block.firstInvalidLine ||
		block.textLineCreationResult != TextLineCreationResult.COMPLETE;
};

internal function getValidLines(block:TextBlock):Array {
	// Create an array of the existing still-valid lines.
	const lines:Array = [];
	const lastValidLine:TextLine = getLineBeforeFirstInvalidLine(block);
	var line:TextLine = block.firstLine;
	while(line && line != lastValidLine) {
		lines.push(line);
		line = line.nextLine;
	}
	return lines;
};

internal function getLineBeforeFirstInvalidLine(block:TextBlock):TextLine
{
	var line:TextLine = block.firstInvalidLine;
	while(line) {
		if(line.validity == TextLineValidity.VALID)
			break;
		line = line.previousLine;
	}
	return line;
};

virtualize.as

package trxcllnt.vr
{
	import asx.array.last;
	import asx.array.map;
	import asx.array.tail;
	import asx.fn.callProperty;
	import asx.fn.distribute;
	import asx.fn.sequence;
	
	import flash.display.DisplayObject;
	import flash.geom.Rectangle;
	
	import org.tinytlf.types.Virtualizer;
	
	import raix.interactive.IEnumerable;
	import raix.reactive.IObservable;

	/**
	 * @author ptaylor
	 */
	public function virtualize(unit:Object,
							   updates:IObservable, /*Array<List, Rectangle, Cache>*/
							   selectVisible:Function, /*(List, Rectangle, Cache):<Enumerable>*/
							   expandChild:Function, /*(Unit):IObservable<Array<Unit, DisplayObject>>*/
							   reportUpdate:Function, /*(Rectangle, Cache, Unit, DisplayObject):void*/
							   expandUpdate:Function):IObservable /*(Rectangle, Cache, Unit, Array<DisplayObject>):IObservable<Array<Unit, DisplayObject>>*/
	{
		return updates.mappend(selectVisible).map(tail).
			switchMany(sequence(
				distribute(function(viewport:Rectangle, cache:Virtualizer, enumerable:IEnumerable):IObservable {
					return concatMany(enumerable, expandChild).
						map(distribute(function(unit:Object, child:DisplayObject):Array {
							return [viewport, cache, unit, child];
						}));
				}),
				callProperty('peek', distribute(reportUpdate)),
				callProperty('toArray')
			)).
			map(function(total:Array):Array {
				return [
					last(total)[0],
					last(total)[1],
					unit,
					map(total, callProperty('slice', -2))
				];
			}).
			switchMany(distribute(expandUpdate));
	}
}