adjusted update grouping, clarifications

This commit is contained in:
Mason Houtz
2021-04-14 21:11:28 -07:00
parent 671002873a
commit e041bc2791
5 changed files with 23 additions and 26 deletions
@@ -2,7 +2,7 @@
<template>
<DscProvider :is-root="isRoot" :debounce-period="500" :collection="selectedCollection" v-slot="{ dsc }">
<CollectionContentProvider v-if="dsc" :parent="dsc" :debug="true" v-slot="{ loading, payload, setScrollPos }">
<CollectionContentProvider v-if="dsc" :parent="dsc" :debug="false" v-slot="{ loading, payload, setScrollPos }">
<ExpandedItems
:scope-key="selectedCollection.id"
:get-item-key="(item) => item.type_id"
@@ -4,28 +4,26 @@ import { content$, dscContent$, buildContentId, buildCollectionId } from "./db/o
import { monitorQuery } from "./db/monitorQuery";
import { SearchParams } from "../model/SearchParams";
import { deepEqual } from "deep-equal";
import { hydrate, shareButDie, show } from "utils/observable";
import { hydrate, shareButDie } from "utils/observable";
import { SEEK } from "./enums";
// history contents monitor
export const monitorHistoryContent = (cfg = {}) => {
const { aggregationKeyField = "hid", ...moreCfg } = cfg;
return twoWayMonitor({
...moreCfg,
...cfg,
db$: content$,
buildRequest: buildContentPouchRequest,
aggregationKeyField,
aggregationKeyField: "hid",
});
};
// collection contents monitor
export const monitorCollectionContent = (cfg = {}) => {
const { aggregationKeyField = "element_index", ...moreCfg } = cfg;
return twoWayMonitor({
...moreCfg,
...cfg,
db$: dscContent$,
buildRequest: buildCollectionPouchRequest,
aggregationKeyField,
aggregationKeyField: "element_index",
});
};
@@ -47,7 +45,6 @@ export const twoWayMonitor = (cfg = {}) => (src$) => {
} = cfg;
const input$ = src$.pipe(
show(debug, ([id, filters, hid]) => console.log("twoWayMonitor inputs", id, hid)),
hydrate([undefined, SearchParams]),
shareButDie(1)
);
@@ -56,18 +53,14 @@ export const twoWayMonitor = (cfg = {}) => (src$) => {
const up$ = input$.pipe(
map(buildRequest({ seek: SEEK.ASC, pageSize })),
distinctUntilChanged(deepEqual),
show(debug, (req) => console.log("twoWayMonitor request (up)", req.selector)),
monitorQuery({ db$, inputDebounce, debug, label: "up" }),
show(debug, ({ doc, match }) => console.log("twoWayMonitor response (up)", doc[aggregationKeyField], match)),
);
// current page + next page = 2 pages down
const down$ = input$.pipe(
map(buildRequest({ seek: SEEK.DESC, pageSize })),
distinctUntilChanged(deepEqual),
show(debug, (req) => console.log("twoWayMonitor request (down)", req.selector)),
monitorQuery({ db$, inputDebounce, debug, label: "down" }),
show(debug, ({ doc, match }) => console.log("twoWayMonitor response (down)", doc[aggregationKeyField], match)),
);
const change$ = merge(up$, down$).pipe(
@@ -79,10 +72,6 @@ export const twoWayMonitor = (cfg = {}) => (src$) => {
const key = doc[aggregationKeyField];
return match ? { key, match, doc, ...more } : { key, match, ...more };
}),
show(debug, update => {
const { key, match } = update;
console.log("monitorHistoryContent change", key, match);
}),
);
return change$;
@@ -103,7 +92,7 @@ export const buildContentPouchRequest = (cfg = {}) => (inputs) => {
const targetId = buildContentId({ history_id, hid });
// SEEK.ASC means the top seek, HID > target
const comparator = seek == SEEK.ASC ? "$gt" : "$lte";
const comparator = seek == SEEK.ASC ? "$gte" : "$lte";
// one page above the target, then 2 after to get the current page and some buffer
const pages = seek == SEEK.ASC ? 1 : 2;
@@ -143,7 +132,7 @@ export const buildCollectionPouchRequest = (cfg = {}) => (inputs) => {
const targetId = buildCollectionId({ parent_url, element_index });
// SEEK.ASC means above the target means element_index < target
const comparator = seek == SEEK.ASC ? "$lt" : "$gte";
const comparator = seek == SEEK.ASC ? "$lte" : "$gte";
// 3 pages total, 1 before, the current one, 1 after;
const pages = seek == SEEK.ASC ? 1 : 2;
@@ -87,7 +87,7 @@ export const ContentProvider = {
initStreams() {
console.warn("Override initStreams in ContentProvider");
return { cache$: NEVER, loader$: NEVER, loading$: NEVER, scrolling$: NEVER };
return { payload$: NEVER, loading$: NEVER, scrolling$: NEVER };
},
/**
@@ -1,7 +1,7 @@
// Generalized bi-directional watch and aggregator.
// Looks at cache from starting index, then looks up and down
import { timer, throwError, from } from "rxjs";
import { throwError, from, EMPTY } from "rxjs";
import {
map,
debounceTime,
@@ -11,6 +11,8 @@ import {
mergeMap,
switchMap,
withLatestFrom,
timeoutWith,
ignoreElements,
} from "rxjs/operators";
import { chunk, show, debounceBurst, shareButDie } from "utils/observable";
import { SearchParams } from "../../model/SearchParams";
@@ -66,15 +68,22 @@ export const aggregateCacheUpdates = (monitor, cfg = {}) => (src$) => {
// Monitor generation function is passed in via config so we can use it against different types of queries
const cacheUpdates$ = monitorInputKey$.pipe(
switchMap(monitor),
// show(debug, (change) => console.log("monitor output", change)),
show(debug, (change) => console.log("monitor output", change)),
);
// groups updates by skiplist key and debounces those
// streams individually so we're not pointlessly updating
// when we're just going to change it again shortly
const groupTimeout = () => timer(updateGroupLifetime);
const groupedUpdates$ = cacheUpdates$.pipe(
groupBy(update => update.key, undefined, groupTimeout),
groupBy(
update => update.key,
update => update,
updateByKey$ => updateByKey$.pipe(
timeoutWith(updateGroupLifetime, EMPTY),
ignoreElements()
)
),
// eliminates rapid double updates to same item
mergeMap(updateByKey$ => updateByKey$.pipe(
debounceTime(0.25 * debouncePeriod),
)),
@@ -51,8 +51,7 @@ export function processContentStreams(payloadOperator, sources = {}, settings =
const payload$ = loaderReset$.pipe(
switchMap(([parent, filters]) => pos$.pipe(
show(debug, pos => {
// console.clear();
console.log("----------------------------------------------")
console.clear();
console.log("processContentStreams: pos", JSON.stringify(pos));
}),
payloadOperator({ parent, filters, loading$, ...settings }),