Firebase отложенной загрузки с rxjs

Question

У меня есть следующий код, чтобы получить последние CHAST сообщения:

const pageSize = 10; 
this.notifier = new Subject<any>(); 
let last: Observable<any>; 

let infiniteList = Observable 

    // Use zip to combine the notifier's emissions with the last 
    // child value: 

    .zip(this.notifier, Observable.defer(() => last)) 

    // Use concatMap to emit a page of children into the 
    // composed observable (note that first is used to complete 
    // the inner list): 

    .concatMap(([unused, last]) => this.af.database.list("chat_messages", { 
     query: { 

     // If there is a last value, start at that value but ask 
     // for one more: 

     limitToLast: last ? (pageSize + 1) : pageSize 
     } 
    }) 
    .first() 
) 

    // Use scan to accumulate the page into the infinite list: 

    .scan((acc, list) => { 

    // If this isn't the initial page, the page was started 
    // at the last value, so remove it from the beginning of 
    // the list: 

    if (acc.length > 0) { 
     list.shift(); 
    } 
    return acc.concat(list); 
    }, []) 

    // Use share so that the last observable (see below) doesn't 
    // result in a second subscription: 

    .share(); 

// Each time a page is emitted, map to its last child value so 
// that it can be fed back into the composed infinite list: 

last = infiniteList 
    .map((list) => { 
    list.reverse(); 
    if (list.length === 0) { 
     return null; 
    } 
    return list[list.length - 1].date_published; 
    }) 
    .startWith(null); 

infiniteList.subscribe((list) => { 
    this.chatMessages = list; 
}); 
this.notifier.next(); 

Всякий раз, когда пользователь прокручивает в нижней части списка я this.notifier.next();, чтобы узнать больше из истории чата. 10 сообщений за раз.

Проблема: первые 10 работ великолепны, но когда я прокручиваю нижнюю часть, чтобы читать новые данные, я получаю ту же информацию в обратном порядке. Например, у меня было так:

1- Хорошо

2- хорошо, вы?

10- Привет alL! Как дела?

--- прокрутки к предыдущему 10 message--

12-. Всем привет! Как дела?

13-. Отлично, ты?

14-. Хорошее

Answer 1

Следующая реализация должна работать нормально.

Обратите внимание, что скомпонованный last наблюдаемый использует filter, чтобы прекратить выдавать значения, когда оно достигает конца (или начало, в зависимости от вашей точки зрения), а последний ключ является первым элементом на странице.

Кроме того, страница должна быть обращена вспять - поскольку Firebase всегда использует восходящий порядок - и обратное должно происходить в скомпонованном pages наблюдаемом, и страница должна быть скопирована по мере того, как происходит разворот.

import { Observable } from "rxjs/Observable"; 
import { Subject } from "rxjs/Subject"; 
import rxjs/add/observable/defer"; 
import rxjs/add/observable/zip"; 
import rxjs/add/operator/concatMap"; 
import rxjs/add/operator/filter"; 
import rxjs/add/operator/first"; 
import rxjs/add/operator/map"; 
import rxjs/add/operator/scan"; 
import rxjs/add/operator/share"; 
import rxjs/add/operator/startWith"; 

const pageSize = 100; 
let notifier = new Subject<any>(); 
let last: Observable<any>; 

let pages = Observable 

    // Use zip to combine the notifier's emissions with the last 
    // child value: 

    .zip(notifier, Observable.defer(() => last)) 

    // Use concatMap to emit a page of children into the 
    // composed observable (note that first is used to complete 
    // the inner list): 

    .concatMap(([unused, last], index) => this.af.database 
    .list("chat_messages", { 
     query: { 
     endAt: last, 
     limitToLast: (index === 0) ? pageSize : (pageSize + 1) 
     } 
    }) 
    .first() 
) 

    // Use share so that the last observable (see below) doesn't 
    // result in a second subscription: 

    .share(); 

// Each time a page is emitted, map to its last child value so 
// that it can be fed back into the composed infinite list: 

last = pages 
    .filter((page) => page.length > 0) 
    .map((page) => page[0].$key) 
    .startWith(undefined); 

// Use scan to accumulate the pages into the infinite list. Copy 
// the page - using slice - before reversing it: 

let infiniteList = pages 
    .scan((list, page) => { 
    page = page.slice(); 
    page.reverse(); 
    if (list.length > 0) { 
     page.shift(); 
    } 
    return list.concat(page); 
    }, []); 

infiniteList.subscribe(observer); 

// Each time the notifier emits, another page will be retrieved 
// and added to the infinite list: 

notifier.next(); 
notifier.next(); 
notifier.next(); 

Это будет работать только надежно при использовании orderByKey. Метод startAt Firebase SDK поддерживает необязательный параметр key при заказе по типу, значению или приоритету. Однако метод endAt документирован как поддерживающий только key при заказе по приоритету. Это означает, что пейджинг с использованием endAt и orderByChild не будет надежным - если дочерние значения не уникальны и есть несколько элементов с одним и тем же дочерним значением, пейджинг не всегда будет возможен.

Что касается обновлений в реальном времени для бесконечного списка, это сложнее, поскольку механизм с ограничениями может легко видеть дубликаты и отсутствующие элементы. Тем не менее, это достаточно просто включить дополнения в реальном времени во главе списка:

let notifier = new Subject<any>(); 
let last: Observable<any>; 

let pages = Observable 
    .zip(notifier, Observable.defer(() => last)) 
    .concatMap(([unused, last], index) => this.af.database 
    .list("chat_messages", { 
     query: { 
     endAt: last, 
     limitToLast: (index === 0) ? pageSize : (pageSize + 1) 
     } 
    }) 
    .first() 
) 
    .share(); 

last = pages 
    .filter((page) => page.length > 0) 
    .map((page) => page[0].$key) 
    .startWith(undefined); 

let tail = pages 
    .scan((list, page) => { 
    page = page.slice(); 
    page.reverse(); 
    if (list.length > 0) { 
     page.shift(); 
    } 
    return list.concat(page); 
    }, []); 

// When the first non-empty page is emitted, create an observable 
// that watches items starting at the first item in the first page: 

let head = pages 
    .filter((page) => page.length > 0) 
    .take(1) 
    .switchMap((page) => this.af.database 
    .list("chat_messages", { 
     query: { 
     startAt: page[page.length - 1].$key 
     } 
    }) 
) 
    .map((list) => list.slice(1)); 

// And combine that with the pages: 

let infiniteList = Observable 
    .combineLatest(
    head, 
    tail, 
    (head, tail) => head.concat(tail) 
); 

infiniteList.subscribe(observer); 

notifier.next(); 
notifier.next(); 
notifier.next();