技术背景：Angular (4) 在 TypeScript 中使用 RxJS。但我认为问题是完全独立于技术的。
我有一系列元素。该数组由服务提供，可能来自远程 API(后端)。
元素服务：
private elements: Element[];
private elementsSource = new ReplaySubject<Element[]>(1);
elementsObservable$ = this.elementsSource.asObservable();
组件订阅该可观察对象，并用数据填充 HTML 列表。现在，用户可以独立编辑列表中的每个元素。编辑后，必须保存更改并将其发送到 a) 后端，b) 订阅该元素更改的其他组件。
由于列表元素可能会被编辑并且它们可能"脏"，因此我不想在其他组件中使用 2 向数据绑定立即显示更改。因此，我复制可观察的输出并处理此副本，然后将带有新列表的更新发送到服务。
元素列表组件：
elements: Element[];
constructor(private elementService: ElementService) { }
ngOnInit() {
this.subscribeElements();
}
private subscribeElements() {
this.elementService.getElements().subscribe(elements => this.elements = clone(elements));
}
private onChangeSave() {
this.elementService.updateElements(clone(this.elements));
}
但是元素列表也可以在其他组件中/由其他用户编辑。在这种情况下，ElementService将通过websocket获得新数据的更新。这将通过可观察发送到 ElementListComponent。
现在，如果：
- 用户 A 将开始编辑一个或多个项目，用户 B 将更新一些其他元素，或者
- 用户将开始编辑一个组件中的一个元素 X，然后在另一个组件中编辑元素 Y，或者
- 用户将开始编辑元素 X 和 Y，保存对元素 X 的更改(保存按钮对于列表中的元素是独立的)，
然后，未保存的更改将丢失，因为列表将被新的更新列表替换。
我想过在 subscribe() 中迭代数组并单独更新本地数组中的每个元素 - 添加新元素、删除缺失的元素、更新更改的元素(如果它们现在没有编辑)。这样，对数组的引用就不会改变，脏元素也不会被"重置"。
这真的是最好的方法吗？还是我错过了什么？
好的，我提出了一个解决方案。我不是通过可观察元素提供完整的数组，而是首先为每个元素发出一个事件，然后每个元素创建/更新/删除一个事件。
元素服务：
@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新订阅者将仅获得每个元素的最后一个过去事件，因此它不会对旧版本的元素执行不必要的操作。
非常欢迎任何想法、评论或更好的想法：-)
考虑调解器模式。
中介器定义控制对象集如何交互的对象。
机场的控制塔是调解员的真实例子。每架飞机都与塔台通信，因此塔台拥有所有必要的信息。