SQLiteを使ってRxJavaでチャットUIを動的に更新する

Question

socket.ioを使ってnodejsサーバ（私が書いた）と通信するチャットアプリがあります。複数の人が1対1でお互いにチャットできます。アプリのUIはWhatsAppのようです。 Chat Threadsがあり、UsersはChat Messagesを交換できます。これらはSQLiteデータベースに格納する必要があります。

サービスがオフになっても機能するように、私はサービスでsocket.io接続を開始しました。

新しいメッセージが来て、順序は、このように

• ある場合はサービスが応答を解析し、アプリがオフになっている場合は、通知が作成され、ユーザ
に示されたSQLite DB
• に書き込みます
• アプリがオンで、ユーザーが新しいメッセージが送信された同じチャットスレッドにいる場合は、recyclerViewを更新する必要があります。
• 基本的な感触は、私がRxJavaを使用してSQLiteでCRUD操作を実行使用したい私はXMPP

をsocket.ioを使用していないよ除くのWhatsAppと同じです。 ChatActivity.class

databaseHelper.getDataObservable(String currentThreadName) 
      .subscribe(new Action1<List<ChatMessage>>() { 
       @Override 
       public void call(List<ChatMessage> chatMessages) { 
        for (ChatMessage message : chatMessages){ 
         //Add to list and update recyclerView 
         mAdapter.notifyDatasetChanged(); 
        } 
       } 
      }); 

問題で

SQLiteHelper.Class

public List<ChatMessage> getChatMessagesForThread(String threadName){ 
    List<ChatMessage>list = new ArrayList<>(); 
    String query = "Some Query Here"; 

    SQLiteDatabase db = this.getReadableDatabase(); 
    Cursor c = db.rawQuery(query, null); 

    if (c.moveToFirst()){ 

     do { 
      ChatMessage message = new ChatMessage(); 
      message.setMessage(c.getString((c.getColumnIndex(CHAT_MESSAGES_KEY_MESSAGE)))); 
      message.setChatThread(c.getInt(c.getColumnIndex(CHAT_MESSAGES_KEY_CHAT_THREAD))); 
      message.setUser(c.getString(c.getColumnIndex(CHAT_MESSAGES_KEY_USER))); 
      list.add(message); 

     } while (c.moveToNext()); 

    }else { 
     //No such thread exists. Returns null 
    } 
    c.close(); 
    return list; 
} 

のRxのJava関連の機能

public static <T> Observable<T> makeObservable(final Callable<T> func) { 
     return Observable.create(
       new Observable.OnSubscribe<T>() { 
        @Override 
        public void call(Subscriber<? super T> subscriber) { 
         try { 
          T observed = func.call(); 
          if (observed != null) { // to make defaultIfEmpty work 
           subscriber.onNext(observed); 
          } 
          //TODO: DECIDE IF THIS STAYS OR NOT !!! 
          //subscriber.onCompleted(); 
         } catch (Exception ex) { 
          subscriber.onError(ex); 
         } 
        } 
       }).subscribeOn(Schedulers.io()); 
    } 

@SuppressWarnings("unchecked") 
private Callable<List<ChatMessage>> getData(String threadName) { 
    return new Callable() { 
     public List<ChatMessage> call() { 
      return getChatMessagesForThread(String threadName); 
     } 
    }; 
} 


    public Observable<List<ChatMessage>> getDataObservable(String threadName) { 
     return makeObservable(getData(String threadName)); 
    } 

は、Rxjavaコールは、上記毎回一度だけではなく、新たな実行されますされます書き込み動作が行われる。

この関数は、SQLiteHelperクラスであると私はスクエアSQL Deliteについて読んなく、上記の作業を取得する方法を理解したいき

public long writeMessageToDB(ChatMessage message){ 
    ContentValues values = new ContentValues(); 
    values.put(CHAT_MESSAGES_KEY_MESSAGE, message.getMessage()); 
    values.put(CHAT_MESSAGES_KEY_CHAT_THREAD, message.getChatThread()); 
    values.put(CHAT_MESSAGES_KEY_USER, message.getUser()); 
    SQLiteDatabase db = this.getWritableDatabase(); 
    return db.insert(CHAT_MESSAGES_TABLE_NAME, null, values); 
} 

私のサービスから呼び出されます。助けてください！

Answer 1

はい、毎回観測可能にしたい場合は.defer（）演算子を使用してください。

Observable.defer(()-> Observable.just("some arg")) 
.subscribeOn(Schedulers.io)) 
.map(t -> getDataFromDb()) 
.map(v -> { 
    // handle your result 
}) 
.subscribe(..);