john/dbt
1
Fork 0
Code Activity

Fix tests

Browse Source
This commit is contained in:
John Ahlroos
2018-05-05 16:05:06 +03:00
parent 2ad773bc56
commit 880e159fe8
8 changed files with 91 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -35,11 +35,11 @@ class BlockTransaction implements Serializable {
}
/**
* End the current transaction
* Commit the transaction to database
*
* @return
*/
void end() {
void commit() {
completed = true
}

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

@@ -5,10 +5,12 @@ import com.devsoap.dbt.data.BlockTransaction
import com.fasterxml.jackson.databind.ObjectMapper
import groovy.util.logging.Slf4j
import ratpack.exec.Promise
import ratpack.func.Pair
import ratpack.handling.Context
import ratpack.handling.Handler
import ratpack.http.Status
import ratpack.http.client.HttpClient
import ratpack.http.client.ReceivedResponse
import ratpack.jdbctx.Transaction
import javax.inject.Inject
@@ -41,21 +43,21 @@ class ExecutorHandler implements Handler {
return
}
executeCommands(ds, transaction).then {
transaction.executed = true
// Notify ledger of result
executeCommands(ds, transaction).onError { e ->
log.info("Sending rolled back transaction to ledger")
println mapper.writeValueAsString(transaction)
client.post(config.ledger.remoteUrl.toURI(), { spec ->
spec.body.text(mapper.writeValueAsString(transaction))
}).then {
ctx.error(e)
}
}.then {
log.info("Updating ledger with execution result")
client.post(config.ledger.remoteUrl.toURI(), { spec ->
spec.body.text(mapper.writeValueAsString(transaction))
}).then {
if(it.status != Status.OK) {
log.error("Failed to update ledger with execution result for transaction $transaction.id")
}
ctx.response.send(mapper.writeValueAsString(transaction))
}
// Return transaction with result
ctx.response.send(mapper.writeValueAsString(transaction))
}
}
}
@@ -81,31 +83,31 @@ class ExecutorHandler implements Handler {
def txDs = Transaction.dataSource(ds)
def tx = Transaction.create { ds.connection }
tx.wrap {
Promise.sync {
try {
transaction.queries.each { block ->
try{
try {
transaction.queries.each { block ->
try{
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)
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
} 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
transaction.executed = true
Promise.sync { transaction }
} catch (Exception e) {
log.error("Failed to execute transaction $transaction.id, transaction rolled back", e)
tx.rollback()
transaction.rolledback = true
Promise.error(e)
}
}
}

85
dbt-core/src/main/groovy/com/devsoap/dbt/handlers/LedgerUpdateTransactionHandler.groovy
View File

@@ -5,6 +5,7 @@ import com.devsoap.dbt.data.BlockTransaction
import com.devsoap.dbt.services.LedgerService
import com.fasterxml.jackson.databind.ObjectMapper
import groovy.util.logging.Slf4j
import ratpack.exec.Promise
import ratpack.handling.Context
import ratpack.handling.Handler
import ratpack.http.HttpMethod
@@ -24,45 +25,55 @@ class LedgerUpdateTransactionHandler implements Handler {
@Override
void handle(Context ctx) throws Exception {
ctx.with {
if(ctx.request.method == HttpMethod.POST) {
if(!config.executor.remoteUrl) {
throw new RuntimeException("Executor URL is not set, cannot update transaction")
}
if (ctx.request.method != HttpMethod.POST) {
ctx.next()
return
}
def ledgerService = get(LedgerService)
request.body.then { body ->
def mapper = get(ObjectMapper)
def transaction = mapper.readValue(body.text, BlockTransaction)
log.info("Recieved transaction $transaction.id")
ledgerService.fetchTransaction(transaction.id).then {
if(it.present) {
log.info "Transaction $transaction.id exists, updating transaction"
ledgerService.updateTransaction(transaction).then {
log.info("Transaction $it updated in ledger")
if(transaction.completed && !transaction.executed){
log.info("Sending transaction $transaction.id to executor at $config.executor.remoteUrl")
redirect(config.executor.remoteUrl)
} else {
render(Jackson.json(transaction))
}
}
} else {
log.info("Creating new transaction")
ledgerService.newTransaction(transaction).then {
log.info("Transaction $it added to ledger")
if(transaction.completed && !transaction.executed){
log.info("Sending transaction $transaction.id to executor at $config.executor.remoteUrl")
redirect(config.executor.remoteUrl)
} else {
render(Jackson.json(transaction))
}
}
}
}
}
if (!config.executor.remoteUrl) {
throw new RuntimeException("Executor URL is not set, cannot update transaction")
}
def ledgerService = ctx.get(LedgerService)
ctx.request.body.then { body ->
def mapper = ctx.get(ObjectMapper)
BlockTransaction transaction = mapper.readValue(body.text, BlockTransaction)
log.info("Recieved transaction $transaction.id")
ledgerService.fetchTransaction(transaction.id).then { Optional<BlockTransaction> t ->
t.present ? updateTransaction(ctx, transaction) : newTransaction(ctx, transaction)
}
}
}
private void updateTransaction(Context ctx, BlockTransaction transaction) {
def ledgerService = ctx.get(LedgerService)
log.info "Transaction $transaction.id exists, updating transaction"
ledgerService.updateTransaction(transaction).then {
log.info("Transaction $it updated in ledger")
if (transaction.completed & !(transaction.executed || transaction.rolledback)) {
log.info("Sending transaction $transaction.id to executor at $config.executor.remoteUrl")
ctx.redirect(config.executor.remoteUrl)
} else {
next()
ctx.render(Jackson.json(transaction))
}
}
}
private void newTransaction(Context ctx, BlockTransaction transaction) {
if(transaction.executed || transaction.rolledback) {
log.error("Tried to create a already executed transaction $transaction.id")
throw new IllegalArgumentException("Cannot create a transaction with executed or rolledback status")
}
def ledgerService = ctx.get(LedgerService)
log.info("Creating new transaction")
ledgerService.newTransaction(transaction).then {
log.info("Transaction $it added to ledger")
if(transaction.completed){
log.info("Sending transaction $transaction.id to executor at $config.executor.remoteUrl")
ctx.redirect(config.executor.remoteUrl)
} else {
ctx.render(Jackson.json(transaction))
}
}
}

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

@@ -128,7 +128,7 @@ class TransactionManagerService implements Service {
}
TransactionBuilder complete() {
transaction.end()
transaction.commit()
this
}