Angular сервис с добавлением / удалением элементов с помощью наблюдаемых

Question

Я работаю над сервисом. У него есть Observable, который используется для вывода товарных строк заказа.

Поток с начальными строками заказа (он может быть пустым)

linesOrder$: Observable<Line[]|any> = this.orderService.selectedOrder.pipe(
share(),
map(
  (order: Order) => {
    if (!order.lines) {
      return of([])
    }
    return order.lines || of()
  }
),
tap(data => console.log('data ', JSON.stringify(data)))
);

Затем я пытаюсь ответить при добавлении или удалении новой строки с использованием действий потока:

//  Insert Action stream
  private lineInsertedSubject = new Subject<Line>();
  lineInsertedAction$ = this.lineInsertedSubject.asObservable();
  linesWithAdd$ = merge(
      this.linesOrder$,
      this.lineInsertedAction$
  ).pipe(...);

 addLine(newLine?: Line) {
   this.lineInsertedSubject.next(newLine)
 }

то же самое для удаления строки:

//  Delete Action stream
  private lineDeletedSubject = new Subject<Line>();
  lineDeletedAction$ = this.lineDeletedSubject.asObservable();
  linesWithDelete$: Observable<Line[]> = merge(
    this.linesOrder$,
    this.lineDeletedAction$
  ).pipe(...)

  deleteLine(line: Line) {
    console.info('eliminar linea');
    this.lineDeletedSubject.next(line);
  }

Дело в том, что я очень запутался, потому что я хочу для создания потока Line [], когда

• Я получил строки начального порядка
• При добавлении строки
• при удалении строки

итак, я создал это в сервисе:

updatedLines$ = merge(
    this.linesOrder$,
    this.linesWithAdd$,
    this.linesWithDelete$
  )
  .pipe(
    tap(data => console.log('updated'))
  );

this updatedLines используется в компоненте следующим образом:

lines$ = this.lineService.updatedLines$;

затем в шаблоне:

<div *ngIf="lines$ | async as lines">
  <p class="text-info">Selected lines: {{ lines.length }}</p>
  <ion-card *ngFor="let line of lines">
    <ion-card-header>
      <ion-card-subtitle>{{ line.item_code }}</ion-card-subtitle>
    </ion-card-header>
    <ion-card-content>
      <ion-item>
        <ion-label class="label">Quantity</ion-label>
        <ion-input type="number" step=".1" (keypress)="numberOnlyValidation($event)" value="{{ line.quantity }}"></ion-input>
        <ion-button color="primary" (click)="onDelete(line)">remove</ion-button>
      </ion-item>
    </ion-card-content>
  </ion-card>
</div>

Я не могу получить обновленные данные в шаблоне, когда я добавляю строку, она достигает метода addLine (), но updatedLines $ не передает. То же самое с deleteLine ().

Сначала я попробовал с combLatest, но обнаружил, что он будет генерировать только при испускании 3 потоков, мне нужно испустить значение, если испускается какой-либо из 3 потоков.

Answer 1

Может быть, вам понадобится scan?

Если бы вы могли предоставить stackblitz с вашими c битами вашего кода, я мог бы предоставить подробности с вашим примером.

Без этого , вот мой код с аналогичной целью:

  productsWithCRUD$ = merge(
    this.productsWithCategory$,
    this.productModifiedAction$
  )
    .pipe(
      scan((products: Product[], product: Product) => this.modifyProducts(products, product)),
      shareReplay(1)
    );

Как и вы, я использую слияние, чтобы объединить исходный поток данных с действием, но я использую одно действие вместо отдельных действий. Каждый раз, когда выполняется действие модификации (создание, обновление, удаление), он использует scan для корректировки массива и его повторной генерации.

Метод modifyProducts выглядит так:

  // Modify the array of products
  modifyProducts(products: Product[], product: Product): Product[] {
    if (product.status === StatusCode.Added) {
      // Return a new array from the array of products + new product
      return [
        ...products,
        { ...product, status: StatusCode.Unchanged }
      ];
    }
    if (product.status === StatusCode.Deleted) {
      // Filter out the deleted product
      return products.filter(p => p.id !== product.id);
    }
    if (product.status === StatusCode.Updated) {
      // Return a new array with the updated product replaced
      return products.map(p => p.id === product.id ?
        { ...product, status: StatusCode.Unchanged } : p);
    }
  }

Надеюсь, это поможет.

Полный код можно найти здесь: https://github.com/DeborahK/Angular-RxJS

Он находится в папке APM-WithExtras в product.service. ts файл.

Answer 2

Как уже упоминалось, способ solid обработки состояния с течением времени в rxjs - это использовать сканирование . Существует solid поддерживаемый способ использования этого оператора очень структурированным, поддерживаемым и поддающимся отладке способом:

const { Subject, of, merge } = rxjs;
const { scan, map } = rxjs.operators;

// observable sources
const initialValue$ = of([1, 2, 3]); // simulates linesOrder$
const add$ = new Subject(); // subject to add a line
const delete$ = new Subject(); // subject to delete a line

/*
update functions that are used to update the state (lines)
- functions first call return another function that takes the state (lines)
- add(7)([1,2,3]) will return [1,2,3,7]
- first call is emitted by the observable (add$, delete$)
- second call is made inside the scan
*/
const override = (update) => (lines) => update;
const add = (num) => (lines) => [...lines, num];
const del = (index) => (lines) => [
  ...lines.slice(0, index),
  ...lines.slice(index + 1, lines.length)
];


const result$ = merge(
  initialValue$.pipe(map(override)), // override is called 1. time with no param
  add$.pipe(map(add)), // add is called 1. time with one param: add(num)
  delete$.pipe(map(del)), // del is called 1. time with one param: del(index)
).pipe(
  scan((lines, fn) => fn(lines), []) // depending on what observable emits the function is now emitted 2. time: fn(lines)
);

result$.subscribe(lines => console.log('lines: ', lines))

// Here the add or delete is triggered
add$.next(4);
add$.next(5);
add$.next(6);
delete$.next(0);
delete$.next(3);

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.3/rxjs.umd.min.js"></script>

Расширение различных операций в вашем состоянии очень просто:

const deleteAll$ = new Subject();

const deleteAll = () => (lines) => []; // This function will delete all lines

// use it then inside the merge
merge(
  ...,
  deleteAll$.pipe(map(deleteAll))
);

// call it whenever you want
deleteAll$.next();

Изменение logi c внутри функций:

// Before we just deleted a line by index of array
const del = (index) => (lines) => [
  ...lines.slice(0, index),
  ...lines.slice(index + 1, lines.length)
];

// Now we want to delete all numbers matching to the given param
const del = (num) => (lines) => lines.filter(line => line !== num);