Использование функций groupBy и toList rxjava с sqlbrite

Question

Когда я использую sqlbrite, я не могу использовать функцию rxjava toList() после использования groupBy(). Вот мой код:

QueryObservable query = jetAnywhereDB.createQuery(Item.TABLE, "SELECT * FROM testTable"); 
query.flatMap(q -> { 
    return q.asRows(Item.MAPPER); 
}). 
groupBy(item -> { 
    return Character.toUpperCase(item.name.charAt(0)); 
}). 
subscribe(groupedObservable -> { 
    groupedObservable.subscribe(item -> { 
    Log.d("test", groupedObservable.getKey() + ": " + item.name); 
    }); 
}); 

Это прекрасно работает и записывает все детали, как они получают излучается, но если я изменить его

groupedObservable.toList().subscribe(items -> { 
    Log.d("test", groupedObservable.getKey() + ": " + items); 
    }); 

ли я не использовать ToList() правильно? Может ли это быть из-за того, что поток sqlbrite не заканчивается, поэтому rxjava ждет, пока больше элементов не испустит, прежде чем завершить toList()?

Answer 1

groupBy - особый вид оператора, который требует от вас подписаться на возвращенный GroupedObservable, иначе он не будет запрашивать больше данных и будет висеть. В первом примере вы делаете это, но во втором, вы начинаете накапливать группы, не подписываясь на них.

Что вы можете сделать, это flatMap пинг группы в паре ключ и список, поэтому он не будет висеть:

.groupBy(t -> ...) 
.flatMap(g -> g.toList().map(lst -> Pair.of(g.getKey(), lst))) 
.subscribe(...) 

Я не знаком с SQLBrite, но это правда не-завершения источник с применяемым к нему toList никогда не испускает ничего (независимо от того, есть ли groupBy или нет).