blob: b7ba963fe205a896ead8eca6095960a960b95cba [file] [log] [blame]
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
using System;
using System.IO;
using System.Threading;
using Apache.NMS.ActiveMQ.Transactions;
using Apache.NMS.ActiveMQ.Transport;
using Apache.NMS.ActiveMQ.Transport.Tcp;
using NUnit.Framework;
namespace Apache.NMS.ActiveMQ.Test
{
[TestFixture]
[Category("Manual")]
class DtcProducerTransactionsTest : DtcTransactionsTestSupport
{
[SetUp]
public override void SetUp()
{
base.SetUp();
this.factory = new NetTxConnectionFactory(ReplaceEnvVar(connectionURI));
this.factory.ConfiguredResourceManagerId = Guid.NewGuid().ToString();
}
[Test]
public void TestRecoverAfterFailOnTransactionCommit()
{
// Test initialize - Fills in DB with data to send.
PrepareDatabase();
using (INetTxConnection connection = factory.CreateNetTxConnection())
{
connection.ExceptionListener += this.OnException;
connection.Start();
ITransport transport = (connection as Connection).ITransport;
TcpFaultyTransport tcpFaulty = transport.Narrow(typeof(TcpFaultyTransport)) as TcpFaultyTransport;
Assert.IsNotNull(tcpFaulty);
tcpFaulty.OnewayCommandPreProcessor += this.FailOnCommitTransportHook;
ReadFromDbAndProduceToQueueWithCommit(connection);
Thread.Sleep(1000);
}
// transaction should not have been commited
VerifyNoMessagesInQueueNoRecovery();
// verify sql server has commited the transaction
VerifyDatabaseTableIsEmpty();
// check messages are present in the queue
NetTxTransactionContext.ResetDtcRecovery();
VerifyBrokerQueueCount();
}
[Test]
public void TestRecoverAfterFailOnTransactionPostCommitSend()
{
// Test initialize - Fills in DB with data to send.
PrepareDatabase();
using (INetTxConnection connection = factory.CreateNetTxConnection())
{
connection.ExceptionListener += this.OnException;
connection.Start();
ITransport transport = (connection as Connection).ITransport;
TcpFaultyTransport tcpFaulty = transport.Narrow(typeof(TcpFaultyTransport)) as TcpFaultyTransport;
Assert.IsNotNull(tcpFaulty);
tcpFaulty.OnewayCommandPostProcessor += this.FailOnCommitTransportHook;
ReadFromDbAndProduceToQueueWithCommit(connection);
Thread.Sleep(1000);
}
// transaction should have been commited
VerifyBrokerQueueCountNoRecovery();
// verify sql server has commited the transaction
VerifyDatabaseTableIsEmpty();
// check messages are present in the queue
VerifyBrokerQueueCount();
}
[Test]
public void TestNoRecoverAfterFailOnTransactionWhenLogDeleted()
{
// Test initialize - Fills in DB with data to send.
PrepareDatabase();
NetTxConnectionFactory netTxFactory = factory;
RecoveryFileLogger logger = netTxFactory.RecoveryPolicy.RecoveryLogger as RecoveryFileLogger;
string logDirectory = logger.Location;
using (INetTxConnection connection = factory.CreateNetTxConnection())
{
connection.ExceptionListener += this.OnException;
connection.Start();
ITransport transport = (connection as Connection).ITransport;
TcpFaultyTransport tcpFaulty = transport.Narrow(typeof(TcpFaultyTransport)) as TcpFaultyTransport;
Assert.IsNotNull(tcpFaulty);
tcpFaulty.OnewayCommandPreProcessor += this.FailOnCommitTransportHook;
ReadFromDbAndProduceToQueueWithCommit(connection);
Thread.Sleep(2000);
}
// transaction should not have been commited
VerifyNoMessagesInQueueNoRecovery();
// delete all recovery files
foreach (string file in Directory.GetFiles(logDirectory, "*.bin"))
{
File.Delete(file);
}
// verify sql server has commited the transaction
VerifyDatabaseTableIsEmpty();
// check messages are NOT present in the queue bacause recovery file has been deleted
VerifyNoMessagesInQueue();
}
[Test]
public void TestNoRecoverAfterFailOnTransactionWhenLogWriteFails()
{
// Test initialize - Fills in DB with data to send.
PrepareDatabase();
string newConnectionUri =
connectionURI + "?nms.RecoveryPolicy.RecoveryLoggerType=harness" +
"&nms.configuredResourceManagerId=" +
factory.ConfiguredResourceManagerId;
factory = new NetTxConnectionFactory(ReplaceEnvVar(newConnectionUri));
using (INetTxConnection connection = factory.CreateNetTxConnection())
{
IRecoveryLogger logger = (connection as NetTxConnection).RecoveryPolicy.RecoveryLogger;
Assert.IsNotNull(logger);
RecoveryLoggerHarness harness = logger as RecoveryLoggerHarness;
Assert.IsNotNull(harness);
harness.PreLogRecoveryInfoEvent += FailOnPreLogRecoveryHook;
connection.ExceptionListener += this.OnException;
connection.Start();
ReadFromDbAndProduceToQueueWithCommit(connection);
Thread.Sleep(2000);
}
// verify sql server has not commited the transaction
VerifyDatabaseTableIsFull();
// check messages are not present in the queue
VerifyNoMessagesInQueue();
}
[Test]
public void TestRecoverAfterFailOnTransactionBeforePrepareSent()
{
// Test initialize - Fills in DB with data to send.
PrepareDatabase();
using (INetTxConnection connection = factory.CreateNetTxConnection())
{
connection.ExceptionListener += this.OnException;
connection.Start();
ITransport transport = (connection as Connection).ITransport;
TcpFaultyTransport tcpFaulty = transport.Narrow(typeof(TcpFaultyTransport)) as TcpFaultyTransport;
Assert.IsNotNull(tcpFaulty);
tcpFaulty.OnewayCommandPreProcessor += this.FailOnPrepareTransportHook;
ReadFromDbAndProduceToQueueWithCommit(connection);
Thread.Sleep(2000);
}
// verify sql server has not commited the transaction
VerifyDatabaseTableIsFull();
// check messages are not present in the queue
VerifyNoMessagesInQueue();
}
[Test]
public void TestRecoverAfterFailOnTransactionDuringPrepareSend()
{
// Test initialize - Fills in DB with data to send.
PrepareDatabase();
using (INetTxConnection connection = factory.CreateNetTxConnection())
{
connection.ExceptionListener += this.OnException;
connection.Start();
ITransport transport = (connection as Connection).ITransport;
TcpFaultyTransport tcpFaulty = transport.Narrow(typeof(TcpFaultyTransport)) as TcpFaultyTransport;
Assert.IsNotNull(tcpFaulty);
tcpFaulty.OnewayCommandPostProcessor += this.FailOnPrepareTransportHook;
ReadFromDbAndProduceToQueueWithCommit(connection);
Thread.Sleep(2000);
}
// verify sql server has commited the transaction
VerifyDatabaseTableIsFull();
// check messages are present in the queue
VerifyNoMessagesInQueue();
}
[Test]
public void TestRecoverAfterTransactionScopeAborted()
{
// Test initialize - Fills in DB with data to send.
PrepareDatabase();
using (INetTxConnection connection = factory.CreateNetTxConnection())
{
connection.ExceptionListener += this.OnException;
connection.Start();
ReadFromDbAndProduceToQueueWithScopeAborted(connection);
Thread.Sleep(2000);
}
// verify sql server has NOT commited the transaction
VerifyDatabaseTableIsFull();
// check messages are NOT present in the queue
VerifyNoMessagesInQueue();
}
[Test]
public void TestRecoverAfterRollbackFailWhenScopeAborted()
{
// Test initialize - Fills in DB with data to send.
PrepareDatabase();
using (INetTxConnection connection = factory.CreateNetTxConnection())
{
connection.ExceptionListener += this.OnException;
connection.Start();
ITransport transport = (connection as Connection).ITransport;
TcpFaultyTransport tcpFaulty = transport.Narrow(typeof(TcpFaultyTransport)) as TcpFaultyTransport;
Assert.IsNotNull(tcpFaulty);
tcpFaulty.OnewayCommandPreProcessor += this.FailOnRollbackTransportHook;
ReadFromDbAndProduceToQueueWithScopeAborted(connection);
Thread.Sleep(2000);
}
// verify sql server has commited the transaction
VerifyDatabaseTableIsFull();
// before recovering, messages should NOT be present in the queue
VerifyNoMessagesInQueueNoRecovery();
// check messages are not present in the queue after recover
VerifyNoMessagesInQueue();
}
[Test]
public void TestIterativeTransactedProduceWithDBDelete()
{
using (INetTxConnection connection = factory.CreateNetTxConnection())
{
connection.ExceptionListener += this.OnException;
connection.Start();
PrepareDatabase();
ReadFromDbAndProduceToQueueWithCommit(connection);
PrepareDatabase();
ReadFromDbAndProduceToQueueWithCommit(connection);
PrepareDatabase();
ReadFromDbAndProduceToQueueWithCommit(connection);
PrepareDatabase();
ReadFromDbAndProduceToQueueWithCommit(connection);
PrepareDatabase();
ReadFromDbAndProduceToQueueWithCommit(connection);
}
// verify sql server has commited the transaction
VerifyDatabaseTableIsEmpty();
// check messages are present in the queue
VerifyBrokerQueueCount(MSG_COUNT * 5);
}
}
}