DeferOperation

I have been working on RxC++ for a few months now. I have added Schedulers and lots of operators. Now I am trying to apply Rx to a C++ Windows Store app.

I was told that Deferrals did not compose well with Rx. This motivated me to design the DeferOperation operator.

DeferOperation demonstrates how Rx can be extended by a few engineers to provide services for many. Another way to say this is that Rx allows Async Algorithms to be written once, carefully, so that many others can re-use them.

These are all references to the Windows 8.1 Accelerometer sample I am modifying.

Usage (App.Xaml.Cpp):

	auto ct = std::make_shared<rx::CurrentThreadScheduler>();
	typedef rxrt::EventPattern<Platform::Object^, SuspendingEventArgs^> SuspendingEventPattern;
	rx::from(suspending)
	.chain<rxrt::defer_operation>(
	[](SuspendingEventPattern ep)
	{
	// defer this operation
	return ep.EventArgs()->SuspendingOperation;
	},
	[](rxrt::OperationPattern<SuspendingOperation^>, SuspendingEventPattern)
	{
	// do this while the operation is deferred
	return SuspensionManager::ReactiveSave();
	},
	ct)
	.publish()
	.connect_forever();

view raw save on suspending hosted with ❤ by GitHub

DeferOperation defaults to the ui thread by reading the context off of the Window, but when the suspending event is fired the Window is not valid, so this uses the current thread scheduler explicitly. publish and connect_forever ensure that the work is done even if nothing subscribes the result.

The implementation follows.

The core of the DeferOperation operator is (cpprx/rx-winrt.hpp):

	// must take the deferral early while the event is still on the stack.
	auto op = make_operation_pattern(sop(t));
	typedef decltype(op) OP;
	return Using(
	// resource factory
	[=]()
	{
	return op;
	},
	// observable factory
	[sob, t](OP op)
	{
	return sob(op, t);
	});

view raw core of deferoperation hosted with ❤ by GitHub

sop is the first lambda from the usage (cpprx/rx-winrt.hpp):

	[](SuspendingEventPattern ep)
	{
	// defer this operation
	return ep.EventArgs()->SuspendingOperation;
	}

view raw sop hosted with ❤ by GitHub

sob is the second lambda from the usage (the OperationPattern is passed so that the SuspendingOperation and the Deferral are available to the sequence if it wants to see the Deadline or Complete early, for example) (cpprx/rx-winrt.hpp)

	[](rxrt::OperationPattern<SuspendingOperation^>, SuspendingEventPattern)
	{
	// do this while the operation is deferred
	return SuspensionManager::ReactiveSave();
	}

view raw sob hosted with ❤ by GitHub

Using will retrieve the resource (OperationPattern), subscribe to the result of sob and Dispose the resource when the subscription completes or errors.

OperationPattern provides the abstraction needed by Using. It will call get and complete the deferral (cpprx/rx-winrt.hpp):

	template <class O>
	struct OperationPattern
	{
	typedef decltype(((O)nullptr)->GetDeferral()) D;
	OperationPattern(O operation) :
	operation(operation),
	deferral(operation->GetDeferral())
	{
	}
	O Operation() const {
	return operation;
	};
	D Deferral() const {
	return deferral;
	};
	void Dispose()
	{
	deferral->Complete();
	}
	operator Disposable() const
	{
	// make sure to capture state and not 'this'.
	// usage means that 'this' will usualy be destructed
	// immediately
	auto local = deferral;
	return Disposable([local]{
	local->Complete();
	});
	}
	private:
	O operation;
	D deferral;
	};
	template <class O>
	OperationPattern<O> make_operation_pattern(O o)
	{
	return OperationPattern<O>(std::move(o));
	}

view raw operationpattern hosted with ❤ by GitHub