A function to decorate some stock AngularJS 1.x directives to accept RxJS Observables. Finished product of https://barrysimpson.net/posts/rx-directive-decorators and https://barrysimpson.net/posts/rx-directive-decorators-update1

rxDecorateDirective.js

'use strict';

function rxDecorateDirective($provide, directiveName) {
	// Duck-typing function lifted from the rx.js source.
	function isObservable(obj) {
		return obj && typeof obj.subscribe === 'function';
	}

	$provide.decorator(directiveName + 'Directive', ['$delegate', 'rx', function($delegate, rx) {
		var directiveConfig = $delegate[0];

		// Save the original compile function for delegation later.
		var originalCompileFn = directiveConfig.compile;

		directiveConfig.compile = function() {
			// Run the original compile function with the arguments we were given in case it does
			// something important. It also returns the linking function we'll delegate to below.
			var originalLinkFn = originalCompileFn.apply(directiveConfig, arguments);

			return function postLink(scope, iElement, iAttrs) {
				var originalScope = scope;
				var linkArgs = arguments;

				// Expression in the directive's attribute value we need to watch.
				var attrExpression = iAttrs[directiveName];

				// If the current scope property value is null or undefined, watch for the first
				// value that isn't, in case it's an Observable.
				originalScope.$toObservable(attrExpression)
					.pluck('newValue')
					.skipWhile(function isUndefinedOrNull(value) {
						return typeof value === 'undefined' || value === null;
					})
					.first()
					.subscribe(function onFirstUsefulValue(firstValue) {
						// If we're bound to an Observable, subscribe to it.
						if (isObservable(firstValue)) {
							// The property name we want the original directive linking function to
							// set a watcher on.
							var RX_VALUE_SCOPE_PROPERTY = '$$rxValue';

							// Get a shared version of the value stream we're bound to, since we'll
							// be making multiple subscriptions.
							var valueStream = firstValue.share();

							// Create a new isolate scope to pass to the original linking function.
							var isolateScope = originalScope.$new(true);

							// Replace `originalScope` with `isolateScope` in the link function
							// args, used when calling the original link function with `apply()`
							// below.
							linkArgs[0] = isolateScope;

							// Replace the original directive name attribute value with the
							// property name we want the original directive to watch instead.
							iAttrs.$set(directiveName, RX_VALUE_SCOPE_PROPERTY);

							// Invoke the original linking function with the modified link args we
							// prepared earlier.
							originalLinkFn.apply(directiveConfig, linkArgs);

							// Subscribe to the Observable, updating the child scope property the
							// original linking function code is watching whenever we receive a new
							// value.
							var valueStreamDisposable = valueStream
								.safeApply(isolateScope, function onNextValue(value) {
									isolateScope[RX_VALUE_SCOPE_PROPERTY] = value;
								})
								.subscribe();

							// Our subscription should not live longer than the scope.
							originalScope.$on('$destroy', function() {
								valueStreamDisposable.dispose();
							});
						} else {
							// Else the directive is not bound to an Observable, so we call the
							// original linking function with its original args.
							originalLinkFn.apply(directiveConfig, linkArgs);
						}
					});
			};
		};

		return $delegate;
	}]);
}