Es sind zwei unterschiedliche Datenbanksysteme im Einsatz und die Programmlogik sollte für die richtige Transaktionabwicklung sorgen.
Daraus ergeben sich zwei Transaktion Fällen.
1. Innere Transaktion.
open TransactionM.Type
open TransactionM.Builder
open TransactionM.Helpers
open System.Data
open System.Data.SqlClient
open System.Threading.Tasks
type Params = {conn : SqlConnection; statements : string list; simulateCancel : bool}
type Either<'a,'b> =
| Result of 'a
| Fail of 'b
let dbCommit (tr:SqlTransaction) =
tr.Commit()
tr
let dbRollback (tr:SqlTransaction) =
tr.Rollback()
tr
let beginDBTransaction (cnn:SqlConnection) cancel =
match cancel with
| false -> cnn.BeginTransaction()
| true -> failwith "error beginDBTransaction"
//check : string -> string -> string -> string -> seq<'a * 'b>
// select data from table to the seq.
let inline check dataSource db userId pswd = seq{
let connStr =
new SqlConnectionStringBuilder(DataSource = dataSource,
InitialCatalog = db, UserID = userId, Password = pswd)
use conn = new SqlConnection(connStr.ConnectionString)
conn.Open()
use comm = new SqlCommand("SELECT Value1, Value2 FROM TEST_TABLE", conn)
use reader = comm.ExecuteReader()
while reader.Read() do
yield unbox reader.["Value1"], unbox reader.["Value2"] }
let inline execNonQuery conn tr s=
use comm = new SqlCommand(s, conn, tr)
let res = comm.ExecuteNonQuery()
res
let inline deleteAll dataSource db userId pswd =
let connStr =
new SqlConnectionStringBuilder(DataSource = dataSource,
InitialCatalog = db, UserID = userId, Password = pswd)
use conn = new SqlConnection(connStr.ConnectionString)
conn.Open()
use comm = new SqlCommand("DELETE FROM TEST_TABLE", conn)
comm.ExecuteNonQuery()
//createTransaction: Params -> TransactionM<'a,'b,TransactionState<string,SqlTransaction>>
let createTransaction p =
transaction {
try
let tr = beginDBTransaction p.conn p.simulateCancel
try
match Seq.forall (((<) 0) << execNonQuery p.conn tr) p.statements with
| true ->
return Commit tr
| false ->
return Rollback tr
with
| e ->
printfn " catch %A" (e.Message)
return Rollback tr
with
| e ->
printfn " catch %A" (e.Message)
return Abort (Some e.Message)
}
//createTransactionWithResult: Params -> TransactionM<'a,'b,TransactionState<string,(SqlTransaction * int)>>
let createTransactionWithResult p = transaction {
try
let tr = beginDBTransaction p.conn p.simulateCancel
try
let l = List.map (execNonQuery p.conn tr) p.statements
match Seq.forall ((<) 0) l with
| true ->
return Commit (tr, List.rev l |> List.head)
| _ ->
return Rollback (tr, 0)
with
| e ->
printfn " catch %A" (e.Message)
return Rollback (tr, 0)
with
| e ->
printfn " catch %A" (e.Message)
return Abort (Some e.Message)
}
//createOuterTransaction:
// TransactionM<'a,'b,TransactionState<'c,('d * 'e)>> ->
// Params ->
// TransactionHandle<'a,'b,TransactionState<string,(Either<SqlTransaction,'f> * Either<'d,'c option>)>> ->
// TransactionM<'a,'b,('g -> 'g)>
let createOuterTransaction inner parms handle =
transaction
{
try
//begin outer DB transaction.
let outer = (beginDBTransaction parms.conn parms.simulateCancel)
try
// start statement from outer transaction.
match execNonQuery parms.conn outer (parms.statements.Item 0) with
| r when r > 0 ->
//get inner transaction monad.
let! innerTransactionState = inner
match innerTransactionState with
| Commit (innerTransaction, result) ->
// final statement from outer transaction.
match execNonQuery parms.conn outer ((parms.statements.Item 1).Replace("$$", result.ToString())) with
| r when r > 0 ->
//commit outer and inner transaction.
return! commit handle (Result outer, Result innerTransaction)
| _ ->
return! rollback handle (Result outer, Result innerTransaction)
| Abort m ->
return! rollback handle (Result outer, Fail m)
| Rollback (innerTransaction, _) ->
return! rollback handle (Result outer, Result innerTransaction)
| _ ->
return! rollback handle (Result outer, Fail None)
| _ ->
return! rollback handle (Result outer, Fail None)
with
| e ->
printfn " catch outer %A" (e.Message)
return! rollback handle (Result outer, Fail None)
with
| e ->
printfn " catch %A" (e.Message)
return! abort handle (Some e.Message)
}
//runOuterInnerTransaction : string list -> bool -> string list -> bool -> string
let runOuterInnerTransaction innerStatements innerCancel outeStatements outerCancel =
let connStr =
new SqlConnectionStringBuilder(DataSource = "SOURCE1",
InitialCatalog ="TESTDB",UserID="USER",Password="PWD")
let connStr1 =
new SqlConnectionStringBuilder(DataSource = "SOURCE2",
InitialCatalog ="TESTDB",UserID="USER",Password="PWD")
use con = new SqlConnection(connStr.ConnectionString)
con.Open()
use con1 = new SqlConnection(connStr1.ConnectionString)
con1.Open()
let innerTr = createTransactionWithResult {conn = con ;
statements = innerStatements;
simulateCancel = innerCancel }
let outerTr = beginT (createOuterTransaction innerTr {conn = con1 ;
statements = outeStatements;
simulateCancel = outerCancel })
match runTransactionState outerTr () with
| Commit (Result outer, Result inner)->
dbCommit inner |> ignore
dbCommit outer |> ignore
"commit."
| Rollback (Result outer, Result inner)->
dbRollback inner |> ignore
dbRollback outer |> ignore
"rollback inner and outer."
| Rollback (Result outer, Fail (Some m)) ->
dbRollback outer |> ignore
"rollback outer." + m
| Rollback (Result outer, Fail None) ->
dbRollback outer |> ignore
"rollback outer."
| Rollback (Fail _, Result inner) ->
failwith "rollback inner without outer."
| Abort (Some message) ->
message + " abort outer."
| Abort None ->
"abort outer."
| _ -> failwith "error."
2. Parallele Transaktionen.
//parallelTasksTransaction : Params -> Params -> bool * string
let parallelTasksTransaction p1 p2 =
let task p = Task.Factory.StartNew(fun () -> runTransactionState (createTransaction p) ())
let tasks = [task p1; task p2] |> List.toArray
let result =
Task.Factory.ContinueWhenAll(
tasks,
(fun (ts:Task<TransactionState<string, SqlTransaction>> []) ->
match ts.[0].Result, ts.[1].Result with
| Commit t1, Commit t2 ->
dbCommit t1|>ignore
dbCommit t2|>ignore
true, "commit."
| Commit t1, Rollback t2 ->
dbRollback t1|>ignore
dbRollback t2|>ignore
false, "rollback. (commit task 1, rollback task 2)"
| Rollback t1, Commit t2 ->
dbRollback t1|>ignore
dbRollback t2|>ignore
false, "rollback. (rollback task 1, commit task 2)"
| Rollback t1, Rollback t2 ->
dbRollback t1|>ignore
dbRollback t2|>ignore
false, "rollback. (rollback task 1, rollback task 2)"
| Rollback t1, Abort (Some m) ->
dbRollback t1|>ignore
false, "rollback. (rollback task 1, abort task 2)" + m
| Abort (Some m), Rollback t2 ->
dbRollback t2|>ignore
false, "rollback. (abort task 1, rollback task 2)." + m
| Commit t1, Abort (Some m) ->
dbRollback t1|>ignore
false, "rollback. (Commit task 1, abort task 2)" + m
| Abort (Some m), Commit t2 ->
dbRollback t2|>ignore
false, "rollback. (abort task 1, Commit task 2)." + m
| Abort (Some m1), Abort (Some m2) ->
false, "abort." + m1 + ". " + m2
| _ -> failwith "execution error."))
result.Result
//runParallelTasksTransaction : string list -> bool -> string list -> bool -> bool * string
let runParallelTasksTransaction statements1 cancel1 statements2 cancel2=
let connStr =
new SqlConnectionStringBuilder(DataSource = "SOURCE1",
InitialCatalog ="TESTDB",UserID="USER",Password="PWD")
let connStr1 =
new SqlConnectionStringBuilder(DataSource = "SOURCE2",
InitialCatalog ="TESTDB",UserID="USER",Password="PWD")
use con = new SqlConnection(connStr.ConnectionString)
con.Open()
use con1 = new SqlConnection(connStr1.ConnectionString)
con1.Open()
parallelTasksTransaction {conn = con; statements = statements1; simulateCancel = cancel1}
{conn = con1 ; statements = statements2; simulateCancel = cancel2}Ein Paar Tests. //test : ('a -> 'b -> 'c -> 'd -> 'e) -> 'a -> 'b -> 'c -> 'd -> unit
let test f a b c d =
deleteAll "SOURCE1" "TESTDB" "USER" "PWD" |>ignore
deleteAll "SOURCE2" "TESTDB" "USER" "PWD" |>ignore
let res = f a b c d
let check1 = check "SOURCE1" "TESTDB" "USER" "PWD"
let check2 = check "SOURCE2" "TESTDB" "USER" "PWD"
printfn "result: %A" res
printfn "check data in db1: %A" check1
printfn "check data in db2: %A" check2
printfn "%s" (String.replicate 30 "+")
printfn "%s" (String.replicate 50 "-")
printfn "run inner transaction tests."
printfn "test Commit."
test runOuterInnerTransaction ["INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (2, 'start inner 2') ";
"INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (3, 'start inner 3')";
"INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (2, 'start inner 4')";
"UPDATE TEST_TABLE SET Value1 = 5, Value2 ='end inner 5' WHERE Value1 =2"]
false
["INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (10, 'start outer')";
"INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES ($$, 'end outer. inner updated $$.') " ]
false
printfn "test Rollback 1."
test runOuterInnerTransaction ["INSERT INTO NONE"]
false
["INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (10, 'outer 10')";
"INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (20, 'outer 20') " ]
false
printfn "test Rollback 2."
test runOuterInnerTransaction ["INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (2, 'inner 2') "] false
["INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (10, 'outer 10')";
"INSERT INTO NONE " ] false
// simulate error in BeginTransaction.
printfn "test Abort inner."
test runOuterInnerTransaction ["INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (2, 'inner 2') "] true
["INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (10, 'outer 10')";
"INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (20, 'outer 20') " ] false
// simulate error in BeginTransaction.
printfn "test Abort outer."
test runOuterInnerTransaction ["INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (2, 'start inner 2') ";
"INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (3, 'start inner 3')";
"INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (2, 'start inner 4')";
"UPDATE TEST_TABLE SET Value1 =5, Value2 ='end inner 5' WHERE Value1 =2"]
false
["INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (10, 'start outer')";
"INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES ($$, 'end outer. inner updated $$.') " ]
true
printfn "%s" (String.replicate 50 "-")
printfn "run parallel tasks tests."
printfn "test Commit."
test runParallelTasksTransaction
[for i in 0..100 ->
"INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (" + i.ToString() + ", 'Task 1 (" + i.ToString() + ")')"]
false
[for i in 100..150 ->
"INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (" + i.ToString() + ", 'Task 2 (" + i.ToString() + ")')"]
false
printfn "test Rollback 1."
test runParallelTasksTransaction
["INSERT INTO NONE"]
false
[for i in 100..150 ->
"INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (" + i.ToString() + ", 'Task 2 (" + i.ToString() + ")')"]
false
printfn "test Rollback 2."
test runParallelTasksTransaction
[for i in 100..200 ->
"INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (" + i.ToString() + ", 'Task 1 (" + i.ToString() + ")')"]
false
["INSERT INTO NONE"]
false
// simulate error in BeginTransaction.
printfn "test Abort."
test runParallelTasksTransaction
[for i in 0..100 ->
"INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (" + i.ToString() + ", 'Task 1 (" + i.ToString() + ")')"]
false
[for i in 100..150 ->
"INSERT INTO TEST_TABLE (Value1,Value2 ) VALUES (" + i.ToString() + ", 'Task 2 (" + i.ToString() + ")')"]
true-------------------------------------------------- run inner transaction tests. test Commit. result: "commit." check data in db1: seq [(5, "end inner 5"); (3, "start inner 3"); (5, "end inner 5")] check data in db2: seq [(10, "start outer"); (2, "end outer. inner updated 2.")] ++++++++++++++++++++++++++++++ test Rollback 1. catch "Falsche Syntax in der Nähe von 'NONE'." result: "rollback inner and outer." check data in db1: seq [] check data in db2: seq [] ++++++++++++++++++++++++++++++ test Rollback 2. catch "Falsche Syntax in der Nähe von 'NONE'." result: "rollback inner and outer." check data in db1: seq [] check data in db2: seq [] ++++++++++++++++++++++++++++++ test Abort inner. catch "error beginDBTransaction" result: "rollback outer.error beginDBTransaction" check data in db1: seq [] check data in db2: seq [] ++++++++++++++++++++++++++++++ test Abort outer. catch "error beginDBTransaction" result: "error beginDBTransaction abort outer." check data in db1: seq [] check data in db2: seq [] ++++++++++++++++++++++++++++++ -------------------------------------------------- run parallel tasks tests. test Commit. result: (true, "commit.") check data in db1: seq [(0, "Task 1 (0)"); (1, "Task 1 (1)"); (2, "Task 1 (2)"); (3, "Task 1 (3)"); ...] check data in db2: seq [(100, "Task 2 (100)"); (101, "Task 2 (101)"); (102, "Task 2 (102)"); (103, "Task 2 (103)"); ...] ++++++++++++++++++++++++++++++ test Rollback 1. catch "Falsche Syntax in der Nähe von 'NONE'." result: (false, "rollback. (rollback task 1, commit task 2)") check data in db1: seq [] check data in db2: seq [] ++++++++++++++++++++++++++++++ test Rollback 2. catch "Falsche Syntax in der Nähe von 'NONE'." result: (false, "rollback. (commit task 1, rollback task 2)") check data in db1: seq [] check data in db2: seq [] ++++++++++++++++++++++++++++++ test Abort. catch "error beginDBTransaction" result: (false, "rollback. (Commit task 1, abort task 2)error beginDBTransaction ") check data in db1: seq [] check data in db2: seq [] ++++++++++++++++++++++++++++++
