채팅 UI를 동적으로 업데이트하는 SQLite가있는 RxJava

Question

socket.io를 사용하여 nodejs 서버 (나와 필자)와 통신하는 채팅 앱이 있습니다. 여러 사람이 일대일로 서로 채팅 할 수 있습니다. 앱의 UI는 WhatsApp와 같습니다. Chat Threads이 있으며 Users은 Chat Messages입니다. 이것들은 SQLite db에 저장 될 필요가 있습니다.

서비스가 꺼져 있어도 앱이 작동하는지 확인하기 위해 서비스에서 socket.io 연결을 시작했습니다. 새 메시지가 오면

는, 순서는

• 서비스가 응답을 구문 분석하고 SQLite는 DB 응용 프로그램이 꺼져있는 경우
• 알림이 사용자
로 작성되어 표시됩니다에 기록하여입니다
• 앱이 켜져 있고 사용자가 새 메시지를 보낸 채팅 스레드와 동일한 채팅 스레드에있는 경우 recyclerView을 업데이트해야합니다. 내가 socket.io하지 XMPP 내가 RxJava를 사용 SQLite에 CRUD 작업을 수행 사용할

을 사용하고 제외

기본적 느낌은 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; 
} 

수신 자바 관련 기능

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(..);