Datenbank (SQLite) Transaktion für Downstream-Betreiber mit rxJava

Question

Ich habe einigen Strom Artikel Unter der Annahme:

Observable<Item> stream = ... 

, was ich erreichen möchte?

• Stream hat eine beliebige Anzahl von Operationen. Alle Operationen, bevor eine Transaktion beginnen sollten außerhalb der Transaktion
• Irgendwie Transaktion des Stroms in der Mitte db.beginTransaction()
• Alle Betreiber nach dem Start Transaktion innerhalb der Transaktion ausgeführt werden sollten beginnen laufen haben
• Transaktion im Fall abgeschlossen werden von erfolgreichen Operationen db.setTransactionSuccessful
• Transaktion hat immer db.endTransaction
• Es beiden Schnipsel haben wird groß sein, beendet sein: offen eine Transaktion in Downstream-Aktivitäten für alle Positionen; Transaktion öffnen und schließen für jedes Artikel in dem Strom

//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 ist Anwendungsinstanz beschreibbarer scoped SQLiteDatabase

ich nun eine Lösung. Aber vielleicht hast du irgendwelche Vorschläge für einen saubereren Weg?

Answer 1

1) Für den Fall, wenn alle Elemente in einzelner Transaktion verarbeitet:

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

2) Für den Fall, dass es separate Transaktion für jeden Artikel ist:

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(), 
     () -> {...}) 

Im Allgemeinen würde ich nicht gehen Sie mit dem zweiten Ansatz, da es möglicherweise zu viele (unkontrollierte) gleichzeitige Datenbankverbindungen erfordert.

3) Sie sollten Ihren Code besser umstrukturieren, so dass Sie zuerst alle erforderlichen Informationen sammeln und anschließend die Datenbank in einem Vorgang in kurzer Transaktion aktualisieren. Dies ist, wie ich es tun würde:

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

ich geschaffen habe einen Transformator eine Transaktion für alle Positionen des Stroms zu erreichen:

/** @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(); 
      }); 

Und nannte es:

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

Beachten Sie, dass i Transaktion in .doOnNext starten und jedes Mal überprüfen wenn die Transaktion bereits gestartet wurde, weil es scheint, dass es unmöglich ist, sie nur beim ersten Mal anzurufen.