好的，我想出了一个解决方案。我不是通过 Observable 提供具有 Elements 的完整数组，而是首先为每个 Element 发出一个事件，然后在以后每个 Element 创建/更新/删除时发出一个事件。
元素服务：
@Injectable()
export class ElementService {
private elementsSource = new ReplaySubject<ElementChange>();
/** Observable that returns separate observable for each subscriber with independent copies of emitted items. */
private elementsObservable = Observable.defer(() => this.elementsSource.asObservable().map(x => <ElementChange>clone(x)));
constructor(private logger: Logger) { }
/**
* Returns observable that emits new value on every change (add, update or remove) of every single Element entity.
* Each observer receives independent copy of entity. New observers receives full history of entities - all events emitted from the beginning
* of obserable's work.
*/
getElements(): Observable<ElementChange> {
return this.elementsObservable;
}
createOrUpdateElements(element: Element) {
if (!element.id) { // generate id for new instance
element.id = uniqid();
}
this.elementsSource.next(new ElementChange(element));
}
removeElemenet(element: Element) {
if (element.id) {
this.elementsSource.next(new ElementChange(element, true));
} else {
this.logger.warn("Cannot emit remove event for unknown Element without ID. Element should be first created by method of this service.", element);
}
}
}
/** Model for Element changes emiting by Observable. */
export class ElementChange {
constructor(public data: Element, public remove: boolean = false) { };
}
元素组件：
export class ElementComponent implements OnInit {
private elements = new Array<Element>();
constructor(private elementService: ElementService) { }
ngOnInit() {
this.subscribeElements();
}
private subscribeElements() {
this.elementService.getElements()
.groupBy(change => change.remove)
.subscribe(group => {
if (group.key === false) { // adding / updating
group
.map(change => change.data)
.subscribe(element => {
let idx = this.elements.findIndex(e => e.id === element.id);
if (idx === -1) { // add
this.elements.push(element);
} else { // update
this.elements[idx] = element;
}
});
} else { // removing
group
.map(element => this.elements.findIndex(e => e.id === element.data.id))
.filter(idx => idx !== -1)
.subscribe(idx => {
this.elements.splice(idx, 1);
});
}
});
}
private addAddElementClick() {
this.elements.push(new Element());
}
private onElementSave(element: Element) {
this.elementService.createOrUpdateElements(element);
}
private onElementEditCancel(idx: number) {
this.removeIfLocalOnlyEntity(idx);
}
private onElementRemove(idx: number) {
this.removeIfLocalOnlyEntity(idx) || this.elementService.removeElemenet(this.elements[idx]);
}
/**
* Removes Element from array if it's local only entity, created locally and never sent to service.
* @param idx Index of entity in elements array
* @returns True if entity was removed, false otherwise.
*/
private removeIfLocalOnlyEntity(idx: number): boolean {
if (this.elements[idx].id === undefined) {
this.elements.splice(idx, 1);
return true;
}
return false;
}
}
优点：
- 只要两个用户/组件不修改同一个元素，就可以在不放弃其他更改的情况下对其进行处理（添加、编辑、删除）。
- 为每个观察者克隆元素实例。由于此对象修改在一个组件中不会在另一个组件中可见，直到
ElementService#createOrUpdateElements() 方法被显式调用。
缺点：
- 发射的元素用
remove 标志包装在另一个类中。
-
ElementComponent#subscribeElements() 中的长订阅实现。使用if 子句而不是groupBy() 可能会更短（半行），但这样更易读且易于扩展。
奖金
当新的观察者订阅时，他会得到所有过去的事件。其中一些可能引用同一个元素（例如添加元素，修改它，再次修改它），但我们的ElementComponent 只需要最后一个。所以ReplaySubject可以替换成自定义的ReplayLastDistinctSubject：
/**
* Subject that emits only last distinct instance of each item that was emitted by the source Observable(s),
* regardless of when the observer subscribes. After subscription all new items are normally emitted to the Observer,
* even if they are not distinct. This way on subsciption Observer gets only the latest version of each emitted item so far
* and then gets all new emits.
*
* Besides distinct filtering the ReplayLastDistinctSubject behaves similary to the ReplaySubject.
*
* Method of comparing items can be specified by providing specific key selector.
*/
export class ReplayLastDistinctSubject<T> extends Subject<T> {
private values: T[] = [];
constructor(private keySelector: (value: T) => any = (x) => x) {
super();
}
protected _subscribe(subscriber: Subscriber<T>): Subscription {
const subscription = super._subscribe(subscriber);
if (subscription && !(<ISubscription>subscription).closed) {
const len = this.values.length;
for (let i = 0; i < len && !subscriber.closed; i++) {
subscriber.next(this.values[i]);
}
}
return subscription;
}
next(value: T) {
this.addNewDistinctValue(value);
super.next(value);
}
private addNewDistinctValue(value: T) {
this.values = this.values.filter(x => this.keySelector(x) !== this.keySelector(value));
this.values.push(value);
}
}
使用ReplayLastDistinctSubject 新订阅者将只获得每个元素的最后一个过去事件，因此它不会对旧版本的元素进行不必要的操作。
非常欢迎任何想法、cmets 或更好的想法 :-)