原生实现异步处理利器 —— Observable

本篇带来用原生实现 Observable，一探内部究竟！！

实现步骤解析如下：

1. Observable 应该是一个 Subject 新对象类型；
2. 并且 Subject 对象有subscribe和next两个函数；
3. subscribe 由 observers （观察者）调用，来订阅可观察的数据流；
4. next由 Subject owner 调用，实现推送/发布新数据；
5. Subject owner 应该知道 observers 何时监听数据生效；
6. 另外，Subject owner 也应该知道 observers 何时不再监听了；
7. 再看 observer，它也应该可以随时被取消；
8. 可以定义一个新的对象，Subscription；
9. Subscription 包含 unsubscribe 函数；
10. 每个 observer 想要停止监听来自 Subject 的数据流时，可以调用unsubscribe实现；

有了以上思路之后，我们可以进一步来看代码实现~

代码实现：

Subscription

let Subscription = function(handlerId, unsubscribeNotificationCallback) {
	let self = this;

	self.unsubscribe = () => {
		if(unsubscribeNotificationCallback) {
			unsubscribeNotificationCallback(handlerId);
		}
	};
	
	return self;
};

注意：Subscription 仅在 unsubscribe 被调用时，通知 Subject；

再看 Subject 完整代码实现：

let Subject = function(subscribersStateChangeNotificationCallback) {
	let self = this;
	
	let handlers = {};
	
	Object.defineProperty(self, "subscribersFound", {
		get() {
			let found = false;
			
			for(const prop in handlers) {
				if(handlers.hasOwnProperty(prop)) {
					found = true;
					break;
				}
			}
			
			return found;
		}
	});
	
	Object.defineProperty(self, "subscribersCount", {
		get() {
			let count = 0;
			
			for(const prop in handlers) {
				if(handlers.hasOwnProperty(prop)) {
					count++;
				}
			}
			
			return count;
		}
	});
	
	let unsubscribeNotificationCallback = (handlerId) => {
		if(handlerId && handlerId !== '' && handlers.hasOwnProperty(handlerId)) {
			delete handlers[handlerId];
			
			if(subscribersStateChangeNotificationCallback && !self.subscribersFound) {
				subscribersStateChangeNotificationCallback(false);
			}
		}
	};
	
	self.subscribe = (handler) => {
		let handlerId = createGuid();
		handlers[handlerId] = handler;
		
		if(subscribersStateChangeNotificationCallback && self.subscribersCount === 1) {
			subscribersStateChangeNotificationCallback(true);
		}
		
		return new Subscription(handlerId, unsubscribeNotificationCallback);
	};
	
	self.next = (data) => {
		for(const handlerId in handlers) {
			handlers[handlerId](data);
		}
	};
	
	return self;
};

let createGuid = function() {  
   return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, function(c) {  
      var r = Math.random()*16|0, v = c === 'x' ? r : (r&0x3|0x8);  
      return v.toString(16);  
   });  
};

Subject Owner 实现：

.
.
.
let subscribersStateChangeNotificationCallback = (subscriberFound) => {
	if(!subscriberFound && isNowWatching) {
		stopWatching();
		isNowWatching = false;
	} else if(subscriberFound && !isNowWatching) {
		startWatching();
	}
};

self.data = new Subject(subscribersStateChangeNotificationCallback);
.
.
.
self.data.next(self.snapshot.data);
.
.
.

Observer 实现：

.
.
.
const dashboardServiceSubscription = myDashboardService.data.subscribe((data) => {
	...
});
.
.
.
dashboardServiceSubscription.unsubscribe();
.
.
.

小结：我们可以看到实现关键是 Subject 对象，更重要的是 发布 和 订阅 的过程！当然，也不能忘了 取消订阅 的功能；

发布和订阅模式来处理异步可以忽视掉时间这个维度，就是不用管时间上的先后，就保证了顺序！这一点，在前面一篇函数式编程中也讲过：《XDM，JS如何函数式编程？看这就够了！（六）》 —— 减少时间状态！

不得不说，都是相通的~~ Σ(⊙▽⊙"a

撰文不易，点赞鼓励👍👍👍👍👍👍

我是掘金安东尼，公众号同名，日拱一卒、日掘一金，再会~