Объединить выбросы Observable

Question

Я новичок в RxJava, и что я хочу сделать, это filter() все строки Array и из нового массива с отфильтрованными строками для обновления моего RecyclerView.

Я пробовал:

  Observable.from(suggestionsArray) 
      .filter(new Func1<String, Boolean>() { 
       @Override 
       public Boolean call(String item) { 
        return item.contains(usersTextEntered); 
       } 
      }) 
      .subscribe(new Action1<String>() { 
       @Override 
       public void call(String i) { 
        adapter.addItem(i); 
       } 
      }); 

    adapter.notifyDataSetChanged(); 

где suggestionsArray является массив строк я изначально и как вы можете видеть, я использую filter() фильтровать только те строки, которые содержат определенную строку, а затем обновить адаптер, но что я get is ConcurrentModificationException, и я предполагаю, что это потому, что каждый элемент из фильтра приходит один за другим, поэтому я думаю, что мне нужно что-то до combine всех выбросов до ввода subscriber.

Я искал RxJava wiki для операторов, но все, что я нашел (слияние/объединение/zip и т. Д.), Имеет отношение к двум различным Observables, поэтому я предполагаю, что я должен искать что-то еще, чтобы объединить выбросы, но я не конечно, как его искать.

Answer 1

Пока вы не указать, где находится источник исключения (т.е. трассировки стека), его только гадать. Но чтобы ответить на ваш вопрос о присоединении к предметам в целом, вы можете сделать это с помощью оператора toList(). Он ожидает вызова onComplete и испускает все элементы как одно событие в виде списка.

Answer 2

Дело в том, откуда берутся ConcurrentModificationException? Список? Адаптер?

Если последнее, вы всегда можете перенести в основной поток:

 Observable 
    .from(suggestionsArray) 
    .filter(item -> item.contains(usersTextEntered)) 
    .subscribeOn(AndroidSchedulers.mainThread()) 
    .subscribe(adapter::addItem, err ->{}, adapter::notifyDataSetChanged);