john/dbt
1
Fork 0
Code Activity

Improve demo

Browse Source
This commit is contained in:
John Ahlroos
2018-05-05 09:20:04 +03:00
parent 76512ee32e
commit c2fda1ffaf
6 changed files with 145 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
dbt-core/src/main/groovy/com/devsoap/dbt/data/BlockTransaction.groovy
View File

@@ -1,7 +1,5 @@
package com.devsoap.dbt.data
import com.fasterxml.jackson.annotation.JsonValue
import com.fasterxml.jackson.databind.JsonNode
import groovy.transform.ToString
import java.nio.charset.StandardCharsets
@@ -54,6 +52,7 @@ class BlockTransaction implements Serializable {
long timeStamp
Map<String, List> result
String resultError
Query() {
// For serialization

31
dbt-core/src/main/groovy/com/devsoap/dbt/handlers/ExecutorHandler.groovy
View File

@@ -82,17 +82,28 @@ class ExecutorHandler implements Handler {
def tx = Transaction.create { ds.connection }
tx.wrap {
Promise.sync {
transaction.queries.each { block ->
log.info "Executing $block.query ..."
if(block.query.toLowerCase().startsWith("select")){
log.info('Saving result from Select query')
def result = txDs.connection
.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_UPDATABLE)
.executeQuery(block.query)
block.result = toMap(result)
} else {
txDs.connection.createStatement().execute(block.query)
try {
transaction.queries.each { block ->
try{
def result = txDs.connection
.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_UPDATABLE)
.executeQuery(block.query)
block.result = toMap(result)
log.info "Executing $block.query ..."
if(block.query.toLowerCase().startsWith("select")){
log.info('Saving result from Select query')
} else {
txDs.connection.createStatement().execute(block.query)
}
} catch (Exception e) {
block.resultError = e.message
throw e
}
}
} catch (Exception e) {
log.error("Failed to execute transaction $transaction.id, transaction rolled back", e)
tx.rollback()
transaction.rolledback = true
}
transaction
}

8
dbt-core/src/main/groovy/com/devsoap/dbt/services/TransactionManagerService.groovy
View File

@@ -55,13 +55,13 @@ class TransactionManagerService implements Service {
httpClient.get(config.ledger.remoteUrl.toURI(), { spec ->
spec.headers.add('X-Transaction-Id', transactionId)
}).flatMap { response ->
if(response.status == Status.of(404)) {
throw new RuntimeException("Transaction with id '$transactionId' could not be found")
}
if(response.status != Status.OK) {
throw new RuntimeException("Server returned ${response.statusCode} ${response.status.message}")
throw new RuntimeException("Ledger returned ${response.statusCode} ${response.status.message} for $transactionId")
}
def oldTransaction = mapper.readValue(response.body.text, BlockTransaction)
if(oldTransaction == null) {
throw new RuntimeException("Transaction with id $transactionId could not be found")
}
if(oldTransaction.completed) {
throw new RuntimeException("Cannot modify a completed transaction")
}