데이터베이스 rxJava

Question

다운 스트림 사업자 (SQLite는) 거래 전 항목 중 일부 스트림이 가정 :

Observable<Item> stream = ... 

난 달성하기 위해 노력하고 무엇을?

• 스트림에는 모든 조작이 있습니다. 트랜잭션을 시작하기 전에 모든 작업은 경우에 완료 될 한
• 은 어떻게 든 db.beginTransaction()
• 가 거래를 시작한 후 모든 사업자는
• 거래 트랜잭션 내에서 실행되어야한다 스트림의 중간에 거래를 시작 트랜잭션 외부에서 실행해야합니다 성공적인 작업의 db.setTransactionSuccessful
• 거래는 항상 두 조각 가지고 좋은 것 db.endTransaction
• 종료해야 : 오픈 한 다운 스트림 작업의 모든 항목에 대한 거래;

  //some upstream operators 
  stream.doOnNext(i -> ...) 
      .map(i -> ...) 
      //somehow start transaction here 
      //operator inside transaction. All database changes will be reverted in case error 
      .doOnNext(i -> /*database ops*/) 
      .subscribe() 
  

  PS

스트림의 각 항목에 대한 개폐 트랜잭션 DB는 응용 프로그램이 지금은 해결책을 가지고

쓰기 SQLiteDatabase의 인스턴스를 범위입니다. 하지만 더 깨끗한 방법에 대한 제안이 있습니까?

Answer 1

1) 일반적

class ItemWithTransaction { 
    Item item; 
    Connection conn; // connection associated with this item 
    boolean rollback; 
} 

stream 
    .map(i -> new ItemWithTransaction(i, openTransaction())) 
    .map(i -> i.conn.executeSql(..., i.item.prop1)) 
    . ... 
    .map(i -> { 
     if (...) i.rollback = true; // error related to this item 
     return i; 
    }) 
    . ... 
    .subscribe(i -> { 
      ... 
      if (i.rollback) i.conn.rollback(); 
      else i.conn.commit(); 
      i.conn.close(); 
     }, 
     e -> rollbackAndCloseAllOpenConnections(), 
     () -> {...}) 

I는 않을 것 너무 많은 (제어되지 않은) 동시 데이터베이스 연결이 필요할 수 있으므로 두 번째 방법을 사용하십시오.

3) 필요한 모든 정보를 먼저 수집 한 다음 간단한 트랜잭션으로 한 번에 데이터베이스 업데이트를 수행하는 것이 좋습니다. 이것은 내가 어떻게 할 것인가입니다 :

stream 
    . ... // processing 
    .buffer(...) // collect updates all or in batches 
    .subscribe(Collection<ItemUpdate> batch -> { 
      database.beginTransaction(); 
      try { 
       ... // update multiple items 
       database.setTransactionSuccessful(); 
      } finally { 
       database.endTransaction(); 
      } 
     }, 
     e -> {...}, 
     () -> {...}); 

Answer 2

나는 스트림의 모든 항목에 하나 거래를 달성하기 위해 변압기를 만들었습니다

/** @return transformer which starts SQLite database transaction for each downstream operators, 
* closes transaction in {@link Observable#doOnTerminate}. So transaction will be closed either on successful completion or error of stream. 
* set previously opened SQLite database transaction to completed in {@link Observable#doOnCompleted} call */ 
public <T> Observable.Transformer<T, T> inTransaction() { 
    return observable -> observable 
      .doOnNext(o -> { 
       if (!database.inTransaction()) database.beginTransaction(); 
      }) 
      .doOnCompleted(() -> { 
       if (database.inTransaction()) database.setTransactionSuccessful(); 
      }) 
      .doOnTerminate(() -> { 
       if (database.inTransaction()) database.endTransaction(); 
      }); 

그리고 전화 : 내가 .doOnNext에서 트랜잭션을 시작하고 각 시간을 확인

stream 
    //start transaction here 
    .compose(inTransaction()) 
    .doOnNext(i -> /*database ops*/) 
    .subscribe() 

주 거래가 이미 시작된 경우 처음에는 전화를 걸 수없는 것 같습니다. 각 항목에 대한 별도의 트랜잭션이있을 때

stream 
    .doOnSubscribe(d -> database.beginTransaction()) 
    . ... 
    .subscribe(v -> {...}, 
     e -> database.endTransaction(), 
     () -> { database.setTransactionSuccessful(); database.endTransaction(); }) 

2) 케이스의 경우 : 모든 항목이 단일 트랜잭션에서 처리되는 경우에 대해