module observe;

import vibe.d;
import vibe.core.core;
import vibe.core.concurrency;
import vibe.core.drivers.native;	// had to add this to get rdmd working
import vibe.core.drivers.win32;		// had to add this to get rdmd working
import std.traits;
import std.functional;
import std.algorithm;
import std.range;
import std.stdio;
import core.atomic;

template ElementType(R)
{
	static if (isInputRange!R)
	{
		alias std.range.ElementType!R ElementType;
	} else
		alias ParameterTypeTuple!(R.NextFunc)[0] ElementType;
}
alias ElementType PayloadType;
template isObservable(O)
{
    enum bool isObservable = is(typeof(
    (inout int = 0)
    {
        O o = void;       // can define a observable object
        alias ElementType!O E;
        o.subscribe((scope E e){},(Exception e){},(){});	// has subscribe function
    }));
}
alias void delegate (Exception) ErrorFunc;
alias void delegate () CompletedFunc;

struct Subscription(Observable)
{
	alias void delegate (ElementType!Observable ) NextFunc;
	Observable o;
	NextFunc onNext;
	ErrorFunc onError;
	CompletedFunc onCompleted;
	bool active = true;
	void dispose()
	{
		o.dispose(this);
	}
}
class ObservableRange(R)
{
	alias void delegate (ElementType!R) NextFunc;
	alias Subscription!(ObservableRange!R) Sub;
	private
	{
		Sub[] subscriptions;
		R range;
	}
	this(R r)
	{
		this.range = r;
	}
	Sub subscribe(NextFunc onNext = null, ErrorFunc onError = null, CompletedFunc onCompleted = null)
	{
		auto sub = Sub(this,onNext,onError,onCompleted);
		subscriptions ~= sub;
		if (onNext is null)	// why is this again?
			return sub;
		runTask(
		{
			foreach(e; this.range)
			{
				foreach(s; subscriptions)
				{
					if (s.onNext && s.active)
						s.onNext(e);
				}
				// todo: we can probably do this a little bit more efficient
				// remove inactive subscriptions
				subscriptions = subscriptions.remove!(s => !s.active).array();
			}
			foreach(s; subscriptions)
				if (s.onCompleted && s.active)
					s.onCompleted();
			subscriptions.length = 0;
		});
		return sub;
	}
	void dispose(Sub s)
	{
		// simply mark the subscription in the list as inactive
		auto idx = subscriptions.countUntil(s);
		if (idx > 0)
			subscriptions[idx].active = false;
	}
}
ObservableRange!R observe(R)(R r)
	if (isInputRange!R || isIterable!R)
{
	return new ObservableRange!R(r);
}
alias observe observeRange;
unittest
{
	assert(isObservable!(typeof(iota(1, 10).observe)));
}
class Subject(E)
{
	alias E ElementType;
	alias void delegate (ElementType) NextFunc;
	alias Subscription!(Subject!E) Sub;
	private
	{
		Sub[] subscriptions;
	}
	void onNext(ElementType e)
	{
		foreach(s; subscriptions)
			if (s.onNext && s.active)
				s.onNext(e);
	}
	void onError(Exception e)
	{
		foreach(s; subscriptions)
			if (s.onError && s.active)
				s.onError(e);
	}
	void onCompleted()
	{
		foreach(s; subscriptions)
			if (s.onCompleted && s.active)
				s.onCompleted();
	}
	Sub subscribe(NextFunc onNext = null, ErrorFunc onError = null, CompletedFunc onCompleted = null)
	{
		auto sub = Sub(this,onNext,onError,onCompleted);
		subscriptions ~= sub;
		return sub;
	}
	void dispose(Sub s)
	{
		subscriptions = subscriptions.remove!(s => !s.active).array();
		s.onCompleted();
	}
}
unittest
{
	assert(is(ElementType!(Subject!int) : int));
	assert(isObservable!(Subject!int));
}
unittest
{
	auto s = new Subject!int;
	s.subscribe((i) => assert(i == 1));
	s.onNext(1);
	assert(is(ElementType!(Subject!int) : int));
}
class Observer(E) : Subject!E
{
	alias void delegate(Observer!(E)) Subscriber;
	private
	{
		bool subscribed = false;
		Subscriber subscriber;
	}
	this(Subscriber subscriber)
	{
		this.subscriber = subscriber;
	}
	override Sub subscribe(NextFunc onNext = null, ErrorFunc onError = null, CompletedFunc onCompleted = null)
	{
		auto sub = super.subscribe(onNext,onError,onCompleted);
		if (!subscribed)
		{
			subscribed = true;
			subscriber(this);
		}
		return sub;
	}
}
// here we test some basic UFCS and unaryFun behaviour that we rely on
template foo(alias fun)
{
	auto foo(T)(T n)
	{
		alias func = unaryFun!fun;
		return func(n);
	}
}
unittest
{
	assert(foo!(a => a*a)(4) == 16);
	assert(4.foo!(a => a*a) == 16);
	assert(foo!"a*a"(4) == 16);
	assert(4.foo!"a*a" == 16);
	auto square = (int n) { return n*n; };
	assert(foo!square(4) == 16);
	assert(4.foo!square == 16);
}
template map(alias fun)
{
	auto map(Obs)(Obs observable)
		if (isObservable!Obs)
	{
		alias func = unaryFun!fun;
		alias AppliedReturnType = typeof(func(ElementType!(Obs).init));

		return new Observer!(AppliedReturnType)
			((self){
				observable.subscribe(
					(e){ self.onNext(func(e)); },
					&self.onError,
					&self.onCompleted
				);
			});
	}
}
//alias map oMap;
unittest
{
	auto s = new Subject!int;
	auto p = s.map!(s => "abs");
	p.subscribe((s){ assert(s == "abs");});
	s.onNext(1);
}
unittest
{
	bool completed = false;
	auto s = new Subject!int;
	auto q = s.map!(s => s*s).map!(s => s*2);
	q.subscribe((s) { assert(s == 18);},null,(){completed = true;});
	s.onNext(3);
	s.onCompleted();
	assert(completed == true);
}
unittest
{
	bool completed = false;
	auto s = new Subject!int;
	auto sub = s.subscribe((v) { assert(v == 18);},null,(){completed = true;});
	s.onNext(18);
	sub.dispose();
	assert(completed == true);
}

class FiberSubject(E) : Subject!E
{
	alias void delegate (E) NextFunc;
	private
	{
		Task t;
		bool started = false;
	}
	void join()
	{
		forceTask();
		t.join();
	}
	private void forceTask()
	{
		if (started)
			return;
		t = runTask(
		{
			bool running = true;
			while (running) {
				receive(
					(E e)
					{
						foreach(s; subscriptions)
							if (s.onNext && s.active)
								s.onNext(e);
					},
					(Exception e)
					{
						foreach(s; subscriptions)
							if (s.onError && s.active)
								s.onError(e);
						running = false;
					},
					(bool)
					{
						foreach(s; subscriptions)
							if (s.onCompleted && s.active)
								s.onCompleted();
						running = false;
					}
				);
			}
		});
		started = true;
	}
	override void onNext(E e)
	{
		forceTask();
		t.send(e);
	}
	override void onError(Exception e)
	{
		forceTask();
		t.send(e);
	}
	override void onCompleted()
	{
		forceTask();
		t.send(true);
	}
}

unittest
{
	auto s = new FiberSubject!int;
	s.subscribe((s) { assert(s == 14);},null,(){writeln("completed");exitEventLoop();});
	s.onNext(14);
	s.onCompleted();
	runEventLoop();
}
unittest
{
	auto s = new FiberSubject!int;
	auto q = s.map!(s=>s*s);
	q.subscribe((s) { assert(s == 28*28);},null,(){exitEventLoop();});
	s.onNext(28);
	s.onCompleted();
	runEventLoop();
}

Observer!(Tuple!(ElementType!ObsA,ElementType!ObsB)) zip(ObsA, ObsB)(ObsA a, ObsB b)
	if (isObservable!ObsA && isObservable!ObsB)
{
	auto result = new Observer!(Tuple!(ElementType!ObsA,ElementType!ObsB))
		((self){
			import vibe.core.sync;
			auto event = createManualEvent();
			int emitA = event.emitCount();
			int emitB = emitA;
			ElementType!ObsA objA;
			bool complete = false;
			a.subscribe((ae) 
				{
					objA = ae;
					event.emit();
					emitA = event.wait(emitA + 1);
				},
				(e) { self.onError(e); },	// TODO: unsubscribe from underlying observable
				() { complete = true; event.emit(); self.onCompleted(); }
			);
			b.subscribe((be) 
				{
					emitB = event.wait(emitB) + 1;
					if (!complete)
						self.onNext(Tuple!(ElementType!ObsA,ElementType!ObsB)(objA,be));
					event.emit();
				},
				(e) { self.onError(e); }, // TODO: unsubscribe from underlying observable
				() { event.emit(); self.onCompleted(); }
			);
		});

	return result;
}

unittest
{
	auto s1 = new FiberSubject!int;
	auto s2 = new FiberSubject!string;
	auto p = s1.zip(s2);
	p.subscribe((t) { assert(t[0] == 1); assert(t[1] == "abc"); },null,(){exitEventLoop();});
	s1.onNext(1);
	s2.onNext("abc");
	s2.onNext("abc");
	s1.onNext(1);
	s1.onCompleted();
	s2.onCompleted();
	runEventLoop();
}

unittest
{
	auto range = iota(0, 10);
	auto s1 = new FiberSubject!int;
	auto s = range.observe.zip(s1);
	s.subscribe((t) { assert(t == tuple(0,5)); },null,(){exitEventLoop();});
	s1.onNext(5);
	s1.onCompleted();
	runEventLoop();
}

template filter(alias fun)
{
	alias func = unaryFun!fun;
	auto filter(Obs)(Obs observable)
		if (is(typeof(func(ElementType!(Obs).init)) : bool) && isObservable!Obs)
	{
		return new Observer!(ElementType!Obs)
			((self){
				observable.subscribe
				(
					(i) { if (func(i)) self.onNext(i); },
					&self.onError,
					&self.onCompleted
				);
			});
	}
}
unittest
{
	// somehow this has tasks still running at exit
	auto range = iota(1, 10);
	range
		.observe
		.map!((i){return i*i;})
		.filter!(i=>i==81)
		.subscribe((i) { assert(i == 81); },null,(){exitEventLoop();});
	runEventLoop();
}

unittest
{
	auto range = iota(1, 3);
	auto s = range.observe.filter!(i=>i % 2 == 0);
	s.subscribe((t) { assert(t == 2); },null,(){exitEventLoop();});
	runEventLoop();
}

unittest
{
	auto event = createManualEvent();
	assert(event.wait(40) == 0);
	event.emit();
	assert(event.wait(40) == 1);
}
unittest
{
	auto event = createManualEvent();
	event.emit();
	assert(event.wait(0) == 1);
}
class Fork(E) : Observer!E
{
	private
	{
		shared int legs;
		int size;
		ReturnType!createManualEvent event;
	}
	this(int legs, Subscriber s)
	{
		super(s);
		this.legs = legs;
		size = legs;
		event = createManualEvent();
	}
	override void onCompleted()
	{
		int cnt = event.emitCount();
		while (cnt+size > legs)
			cnt = event.wait(cnt);
		super.onCompleted();
	}
	override void onNext(E e)
	{
		event.wait(legs);
		legs.atomicOp!"-="(1);
		runTask(
			{
				foreach(s; subscriptions)
					if (s.onNext && s.active)
						s.onNext(e);
				legs.atomicOp!"+="(2);
				event.emit();
			});
	}
}

Fork!(ElementType!Obs) fork(Obs)(Obs o, int legs)
	if (isObservable!Obs)
{
	import std.exception;
	enforce(legs > 0, "can't fork without legs");
	return new Fork!(ElementType!Obs)(legs,
		(self){
			o.subscribe(
				&self.onNext,
				&self.onError,
				&self.onCompleted
			);
		});
}


unittest
{
	int cnt = 0;
	iota(0,10).observe.fork(10).subscribe((e){ cnt += e; },null,() {exitEventLoop(); assert(cnt == 45); });
	runEventLoop();
}

Observer!(ElementType!ObsA) chain(ObsA, ObsB)(ObsA a, ObsB b)
	if (isObservable!ObsA && isObservable!ObsA && is (ElementType!ObsA : ElementType!ObsB))
{
	return new Observer!(ElementType!ObsA)
		((self)
		{
			auto event = createManualEvent();
			a.subscribe(
				&self.onNext,
				&self.onError,	// TODO: instead we need to unsubscribe from b, then propagate error
				() { event.emit(); if (event.emitCount() == 2) self.onCompleted(); }
			);
			b.subscribe(
				&self.onNext,
				&self.onError,	// TODO: instead we need to unsubscribe from a, then propagate error
				() { event.emit(); if (event.emitCount() == 2) self.onCompleted(); }
			);
		});
}

unittest
{
	auto s = new FiberSubject!int;
	int cnt = 0;
	iota(0,10).observe.chain(s).subscribe((e){ cnt += e; },null,() {exitEventLoop(); assert(cnt == 76); });
	s.onNext(15);
	s.onNext(16);
	s.onCompleted();
	runEventLoop();
}
class ConcatMap(T, alias func)
{
	alias AppliedReturnType = typeof(func(T.init));
	alias ElementType!(AppliedReturnType) E;
	alias void delegate (E) NextFunc;
	alias void delegate(ConcatMap!(T, func)) SubscribeFunc;
	alias Subscription!(ConcatMap!(T, func)) Sub;
	private
	{
		bool subscribed = false;
		SubscribeFunc subscriber;
		Sub[] subscriptions;
		shared int cnt;
	}
	this(SubscribeFunc sFunc)
	{
		this.subscriber = sFunc;
		this.cnt = 1;
	}
	private void subCompleted()
	{
		if (cnt.atomicOp!"-="(1) != 0)
			return;
		foreach (s; subscriptions)
			if (s.onCompleted)
				s.onCompleted();
		subscriptions.destroy();
	}
	void onNext(T e)
	{
		cnt.atomicOp!"+="(1);
		auto obs = func(e);
		obs.subscribe(null,null,&this.subCompleted);
		foreach (s; subscriptions)
			obs.subscribe(s.onNext,s.onError); // here we need to call the class' onCompleted which should only trigger after all obs.onCompleted are triggered and a call has been made to onCompleted from the underlying subject.
	}
	void onError(Exception e)
	{
		foreach (s; subscriptions)
			if (s.onError)
				s.onError(e);
	}
	void onCompleted()
	{
		subCompleted();
	}
	Sub subscribe(NextFunc onNext = null, ErrorFunc onError = null, CompletedFunc onCompleted = null)
	{
		auto sub = Sub(this,onNext,onError,onCompleted);
		subscriptions ~= sub;
		if (!subscribed)
		{
			subscribed = true;
			subscriber(this);
		}
		return sub;
	}
	void dispose(Sub s)
	{
		// todo: make work
	}
}
template concatMap(alias fun)
{
	alias func = unaryFun!fun;

	auto concatMap(Obs)(Obs o)
		if (isObservable!Obs && isObservable!(typeof(func(ElementType!(Obs).init))))
	{
		return new ConcatMap!(ElementType!Obs,func)(
			(self)
			{
				o.subscribe(
					&self.onNext,
					&self.onError,
					&self.onCompleted);
			});
	}
	auto concatMap(Obs)(Obs o)
		if (isObservable!Obs && isInputRange!(typeof(func(ElementType!(Obs).init))))
	{
		return o.concatMap!(a => func(a).observe())();
	}
}
unittest
{
	auto s = iota(0,5).observe;
	size_t c;
	s.concatMap!(s => iota(0,5).observe() ).subscribe((p){c += p;},null,(){ exitEventLoop(); assert(c == 50); });
	runEventLoop();
}
unittest
{
	auto s = iota(0,5).observe;
	size_t c;
	s.concatMap!(s => iota(0,5) ).subscribe((p){c += p;},null,(){ exitEventLoop(); assert(c == 50); });
	runEventLoop();
}
//class ObservableTimer
//{
//	alias void delegate (int) NextFunc;
//	private
//	{
//		uint delay, period;
//	}
//	this(uint d, uint p)
//	{
//		this.delay = d;
//		this.period = p;
//	}
//	void subscribe(NextFunc onNext = null, ErrorFunc onError = null, CompletedFunc onCompleted = null)
//	{
//		if (onNext is null)
//			return;
//		runTask(
//		{
//			sleep(this.delay);
//			ulong cnt;
//			while (true)
//			{
//				onNext(cnt);
//				cnt++;
//				sleep(period);
//			}
//		});
//	}	
//}
//auto timer(int delay, int period)
//{
//	return new ObservableTimer(delay,period);
//}
template tap(alias fun)
{
	alias func = unaryFun!fun;

	auto tap(Obs)(Obs o)
		if (isObservable!Obs && is(typeof(func(ElementType!(Obs).init)) : void))
	{
		return new Observer!(ElementType!Obs)(
			(self)
			{
				o.subscribe(
					(e){ func(e); self.onNext(e); },
					&self.onError,
					&self.onCompleted);
			});
	}
	//auto tap(Obs)(Obs o)
	//	if (isObservable!Obs && isObservable!(typeof(func(ElementType!(Obs).init))))
	//{
	//	return new Observer!(ElementType!Obs)(
	//		(self)
	//		{
	//			// the o observable might trigger onCompleted before the observable returned by the func completes
	//			// in which case one or more onNext will be called AFTER onCompleted
	//			o.subscribe(
	//				(e){ func(e).subscribe(null,$self.onError,(){self.onNext(e);}); },
	//				&self.onError,
	//				&self.onCompleted);
	//		});
	//}
}
/*unittest
{
	auto s = iota(0,5).observe;
	size_t c;
	s.tap!((i) { c += i; }).subscribe((i){ c+= i; },null,(){ exitEventLoop(); assert(c == 20); });
	runEventLoop();
}*/
//class Catch(Obs) : Subject!(ElementType!Obs)
//{
//	private
//	{
//		bool subscribed = false;
//		Obs o;
//	}
//	this(Obs o)
//	{
//		this.o = o;
//	}
//	override void onError(Exception e)
//	{
//		onCompleted();
//	}
//	override void subscribe(NextFunc onNext = null, ErrorFunc onError = null, CompletedFunc onCompleted = null)
//	{
//		super.subscribe(onNext,onError,onCompleted);
//		if (!subscribed)
//		{
//			o.subscribe(&this.onNext,&this.onError,&this.onCompleted);
//			subscribed = true;
//		}
//	}
//}
//Catch!(Obs) catchException(Obs)(Obs o)
//	if (isObservable!Obs)
//{
//	return new Catch!(Obs)(o);
//}