Transaction support
177 views
Skip to first unread message
Geoff
Aug 31, 2012, 3:01:09 AM8/31/12
to masstrans...@googlegroups.com
Hi all,
What is the current recommendation for rolling back DB transactions in response to exceptions within a consumer?
BeforeConsumingMessage/AfterConsumingMessage don't seem appropriate, as there is no context in AfterConsumingMessage to determine whether an exception occured.
IConsumerFactory a-la https://groups.google.com/d/msg/masstransit-discuss/fUlpujoPxzM/I2sU0vow0bMJ doesn't seem to work either - the exceptions don't bubble up and are instead trapped within the action's execution, as shown in the snippet below.
An exception which occurs inside the yeild results (NOTE 1) never bubbles up to prevent the subsequent code (NOTE 2) from executing, thus this block always commits.
public System.Collections.Generic.IEnumerable<Action<IConsumeContext<TMessage>>> GetConsumer<TMessage>(IConsumeContext<TMessage> context,
MassTransit.Pipeline.InstanceHandlerSelector<T, TMessage> selector) where TMessage : class
{
using (var childContainer = _container.CreateChildContainer())
{
var consumer = childContainer.Resolve<T>();
if (consumer == null)
throw new ConfigurationException(string.Format("Unable to resolve type '{0}' from container: ", typeof(T)));
var session = .. stuff to get nhibernate session ..
try
{
using (var tx = session.BeginTransaction())
{
foreach (var handler in selector(consumer, context))
{
yield return handler; // ** NOTE 1 **
}
// ** NOTE 2 **
try
{
session.Flush();
tx.Commit();
}
catch (Exception e)
{
_log.Error(string.Format("Exception thrown saving to database. Message Type: {0}. ", typeof(T).Name), e);
throw;
}
}
}
finally
{
.. stuff to release nhibernate session ..
}
}
}
}
What would you recommend? I've scoured the group and whilst the quesitons been asked a few times there doesn't seem to be a definitive answer.
We are currently using MT 2.1.410.0 on RabbitMQ 2.8.2
Cheers,
Geoff
Dawid Ciecierski
Aug 31, 2012, 8:24:14 AM8/31/12
to masstrans...@googlegroups.com
Have you had a look at TransactionScope? Not sure how it would behave in your particular case and whether it is able to participate in your distributed workflow, but works for us just fine. Only need to remember to define transactional queues rather than nontransactional ones.
using (var transaction = new TransactionScope())
{
result = handler.Handle(command, principal);
transaction.Complete();
}
(Behind the Handle() there are multiple database calls and message publishing operations.)
Regards,
Dawid
using (var transaction = new TransactionScope())
{
result = handler.Handle(command, principal);
transaction.Complete();
}
(Behind the Handle() there are multiple database calls and message publishing operations.)
Regards,
Dawid
Geoff
Sep 4, 2012, 11:36:05 PM9/4/12
to masstrans...@googlegroups.com
Hi Dawid,
We ended up pushing an IUnitOfWorkFactory down into the consumer via our IOC. It works but I don't really like pushing this level of knowledge down to the consumer, I would have rathered it stay unaware of the transaction boundaries its working within...
As to TransactionScope alone, I couldn't work out where you had that code - short of directly inside the consumer itself. If it was inside a IConsumerFactory<T> implementation's GetConsumer() it would always commit - as the exceptions from consumers are not bubbled up. Perhaps you can post a bigger snippet?
Here's the solution I went with for now:
The classes....
public class NHibernateUnitOfWorkFactory : IUnitOfWorkFactory
{
private readonly ISession _session;
public NHibernateUnitOfWorkFactory(ISession session)
{
_session = session;
}
public IUnitOfWork CreateNew()
{
return new NhibernateUnitOfWork(_session);
}
}
public class NhibernateUnitOfWork : IUnitOfWork, IDisposable
{
private readonly ISession _session;
private ITransaction _transaction;
public NhibernateUnitOfWork(ISession session)
{
_session = session;
_session.FlushMode = FlushMode.Auto;
_transaction = _session.BeginTransaction();
}
public void Complete()
{
// because flushMode is auto, this will automatically commit when disposed
if (!_transaction.IsActive)
throw new InvalidOperationException("No active transaction");
_transaction.Commit();
// start a new transaction
_transaction = _session.BeginTransaction();
}
public void Abort()
{
if (_transaction.IsActive)
_transaction.Rollback();
}
#region IDisposable Members
public void Dispose()
{
Abort();
}
#endregion
}
// Setup of our Unit IOC
container.RegisterInstance(new ManagedHibernateSessionContext());
container.RegisterType<ISession>(new TransientLifetimeManager(),
new InjectionFactory(f => _container.Resolve<ManagedHibernateSessionContext>().GetSession(_sessionFactory)));
container.RegisterType<IUnitOfWorkFactory, NHibernateUnitOfWorkFactory>(new TransientLifetimeManager());
container.RegisterType<IUnitOfWork, NhibernateUnitOfWork>(new TransientLifetimeManager());
// Setup of the bus...
_bus = ServiceBusFactory.New(sbc =>
{
sbc.UseRabbitMqRouting();
.....
sbc.BeforeConsumingMessage(() =>
{
var session =
_container.Resolve<ManagedHibernateSessionContext>().GetSession(
_sessionFactory);
});
sbc.AfterConsumingMessage(() =>
_container.Resolve<ManagedHibernateSessionContext>().ReleaseSession(_sessionFactory)
);
sbc.Subscribe((configurator) =>
{
configurator.Consumer(new UnityConsumerFactory<ConsumerA>(_container));
configurator.Consumer(new UnityConsumerFactory<ConsumerB>(_container));
});
...
});
// And finally, our consumers...
public class ConsumerA : Consumes<SomeMsg>.All, IDisposable {
private readonly IUnitOfWorkFactory _unitOfWorkFactory;
public ConsumerA(SomeDependancy1 dep1, SomeDependancy2 dep2, IUnitOfWorkFactory unitOfWorkFactory){
_unitOfWorkFactory = unitOfWorkFactory;
}
public void Consume(EstimatedTimetableAnnouncement msg){
using (var uow = _unitOfWorkFactory.CreateNew()){
try{
... some stuff ...
uow.Complete();
} catch(Exception ex){
uow.Abort();
..... some error handling, maybe requeue ....
}
}
}
}
Dawid Ciecierski
Sep 6, 2012, 2:54:17 AM9/6/12
to masstrans...@googlegroups.com
Geoff,
We used to have units of work all over the place - both for NHibernate and MassTransit. Because we're trying to loosely follow CQRS principles, we were wrapping our CommandHandlers with NhTransactionalHandler and then MtTransactionalHandlers to make this work. However, the beauty of TransactionScope is that it gets picked up by both NHibernate and MassTransit! With that we get full consistency with little work.
Without further ado, here is our CommandDispatcher (with sanity and security checks omitted for brevity):
public class CommandDispatcher
{
public static TResult Dispatch<TCommand, TResult>(TCommand command)
where TCommand : class, ICommand<TResult>
where TResult : ICommandResult, new()
{
var container = Configuration.DependencyResolver.Current;
TResult result;
try
{
using (var scope = container.BeginLifetimeScope())
{
// resolve command handler
var handler = scope.Resolve<ICommandHandler<TCommand, TResult>>();
if (handler == null)
throw new InvalidOperationException("Handler for command registered, but not found.");
using (var transaction = new TransactionScope())
{
result = handler.Handle(command, principal);
transaction.Complete();
}
}
}
catch (Exception e)
{
result = new TResult { Exception = e };
}
return result;
}
}
Command handlers are registered with Autofac like so:
builder.RegisterAssemblyTypes(Assembly.GetExecutingAssembly())
.Where(t => t.Name.EndsWith("CommandHandler"))
.AsImplementedInterfaces()
.InstancePerDependency();
We then register CommandDispatcher's Dispatch method with MassTransit in some convoluted automatic way (which is beyond the scope of this post ;-) and that's it.
Let me know if you'd like to know more about particular bits or if you find that something doesn't work (possible, the code still isn't in production).
Regards,
Dawid Ciecierski
We used to have units of work all over the place - both for NHibernate and MassTransit. Because we're trying to loosely follow CQRS principles, we were wrapping our CommandHandlers with NhTransactionalHandler and then MtTransactionalHandlers to make this work. However, the beauty of TransactionScope is that it gets picked up by both NHibernate and MassTransit! With that we get full consistency with little work.
Without further ado, here is our CommandDispatcher (with sanity and security checks omitted for brevity):
public class CommandDispatcher
{
public static TResult Dispatch<TCommand, TResult>(TCommand command)
where TCommand : class, ICommand<TResult>
where TResult : ICommandResult, new()
{
var container = Configuration.DependencyResolver.Current;
TResult result;
try
{
using (var scope = container.BeginLifetimeScope())
{
// resolve command handler
var handler = scope.Resolve<ICommandHandler<TCommand, TResult>>();
if (handler == null)
throw new InvalidOperationException("Handler for command registered, but not found.");
using (var transaction = new TransactionScope())
{
result = handler.Handle(command, principal);
transaction.Complete();
}
}
}
{
result = new TResult { Exception = e };
}
return result;
}
}
Command handlers are registered with Autofac like so:
builder.RegisterAssemblyTypes(Assembly.GetExecutingAssembly())
.Where(t => t.Name.EndsWith("CommandHandler"))
.AsImplementedInterfaces()
.InstancePerDependency();
We then register CommandDispatcher's Dispatch method with MassTransit in some convoluted automatic way (which is beyond the scope of this post ;-) and that's it.
Let me know if you'd like to know more about particular bits or if you find that something doesn't work (possible, the code still isn't in production).
Regards,
Dawid Ciecierski
Henrik Feldt
Sep 6, 2012, 9:16:29 AM9/6/12
to masstrans...@googlegroups.com
Serialising all transactions are we?
--
You received this message because you are subscribed to the Google Groups "masstransit-discuss" group.
To post to this group, send email to masstrans...@googlegroups.com.
To unsubscribe from this group, send email to masstransit-dis...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msg/masstransit-discuss/-/_wourkOB-NIJ.
For more options, visit https://groups.google.com/groups/opt_out.
Dawid Ciecierski
Sep 7, 2012, 1:21:44 PM9/7/12
to masstrans...@googlegroups.com
Not sure what you mean there Henrik... serialization as in transforming into eg. JSON or sequentiality?
Regards,
Dawid
Regards,
Dawid
To unsubscribe from this group, send email to masstransit-discuss+unsub...@googlegroups.com.
Henrik Feldt
Sep 10, 2012, 12:17:19 PM9/10/12
to masstrans...@googlegroups.com
No I mean that unless you pass ctor params to the txscope you just created a serializable tx.
To unsubscribe from this group, send email to masstransit-dis...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msg/masstransit-discuss/-/xr553B3U23cJ.
Dawid Ciecierski
Sep 11, 2012, 4:55:55 AM9/11/12
to masstrans...@googlegroups.com
Being rather new to all this I still don't understand what you mean by "serializable tx" and how it impacts what we want TransactionScope to do.
Regards,
Dawid
Regards,
Dawid
To unsubscribe from this group, send email to masstransit-discuss+unsubscribe...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msg/masstransit-discuss/-/_wourkOB-NIJ.
For more options, visit https://groups.google.com/groups/opt_out.
--
You received this message because you are subscribed to the Google Groups "masstransit-discuss" group.
To post to this group, send email to masstrans...@googlegroups.com.
To unsubscribe from this group, send email to masstransit-discuss+unsub...@googlegroups.com.
Chris Patterson
Sep 11, 2012, 8:06:03 AM9/11/12
to masstrans...@googlegroups.com
When you create a TransactionScope and Serializable, the transaction
manager will raise all of your transactions to this level, meaning on
a single process can do anything to your database at one time. A
better isolation level is RepeatableRead, or perhaps ReadComitted,
depending upon your situation. It's a parameter of the
TransactionScope constructor.
Serializable is the slowest isolation level, basically one at a time,
first come first served, just like the DMV.
On Tue, Sep 11, 2012 at 3:55 AM, Dawid Ciecierski
>>>>> masstransit-dis...@googlegroups.com.
manager will raise all of your transactions to this level, meaning on
a single process can do anything to your database at one time. A
better isolation level is RepeatableRead, or perhaps ReadComitted,
depending upon your situation. It's a parameter of the
TransactionScope constructor.
Serializable is the slowest isolation level, basically one at a time,
first come first served, just like the DMV.
On Tue, Sep 11, 2012 at 3:55 AM, Dawid Ciecierski
>>>>> To view this discussion on the web visit
>>>>> https://groups.google.com/d/msg/masstransit-discuss/-/_wourkOB-NIJ.
>>>>> For more options, visit https://groups.google.com/groups/opt_out.
>>>>>
>>>>>
>>>
>>> --
>>> You received this message because you are subscribed to the Google Groups
>>> "masstransit-discuss" group.
>>> To post to this group, send email to masstrans...@googlegroups.com.
>>> To unsubscribe from this group, send email to
>>> masstransit-dis...@googlegroups.com.
>>>>> https://groups.google.com/d/msg/masstransit-discuss/-/_wourkOB-NIJ.
>>>>> For more options, visit https://groups.google.com/groups/opt_out.
>>>>>
>>>>>
>>>
>>> --
>>> You received this message because you are subscribed to the Google Groups
>>> "masstransit-discuss" group.
>>> To post to this group, send email to masstrans...@googlegroups.com.
>>> To unsubscribe from this group, send email to
>>> To view this discussion on the web visit
>>> https://groups.google.com/d/msg/masstransit-discuss/-/xr553B3U23cJ.
>>> For more options, visit https://groups.google.com/groups/opt_out.
>>>
>>>
>
>>> https://groups.google.com/d/msg/masstransit-discuss/-/xr553B3U23cJ.
>>> For more options, visit https://groups.google.com/groups/opt_out.
>>>
>>>
>
> --
> You received this message because you are subscribed to the Google Groups
> "masstransit-discuss" group.
> To post to this group, send email to masstrans...@googlegroups.com.
> To unsubscribe from this group, send email to
> masstransit-dis...@googlegroups.com.
> You received this message because you are subscribed to the Google Groups
> "masstransit-discuss" group.
> To post to this group, send email to masstrans...@googlegroups.com.
> To unsubscribe from this group, send email to
> To view this discussion on the web visit
> https://groups.google.com/d/msg/masstransit-discuss/-/ytf67AXUmukJ.
Henrik Feldt
Sep 11, 2012, 5:15:15 PM9/11/12
to masstrans...@googlegroups.com
What Chris said.
Have a look at these pages on MSDN:
http://msdn.microsoft.com/en-us/library/ms149853.aspx for the c'tor that you can call with a custom IsolationMode enum, and
http://msdn.microsoft.com/en-us/library/system.transactions.transactionoptions.aspx for the options struct that you pass.
Regards,
Henrik
Have a look at these pages on MSDN:
http://msdn.microsoft.com/en-us/library/ms149853.aspx for the c'tor that you can call with a custom IsolationMode enum, and
http://msdn.microsoft.com/en-us/library/system.transactions.transactionoptions.aspx for the options struct that you pass.
Regards,
Henrik
Dawid Ciecierski
Sep 21, 2012, 9:25:09 AM9/21/12
to masstrans...@googlegroups.com
Chris / Henrik,
Thank you for raising my awareness of this, somehow it never ocurred to me to check all other availbable ctors for TransactionScope and therefore didn't realise the implications of using a parameterless one. RepeatableRead does indeed seem a more reasonable option, at least the way our system works.
Thanks!
Dawid Ciecierski
Thank you for raising my awareness of this, somehow it never ocurred to me to check all other availbable ctors for TransactionScope and therefore didn't realise the implications of using a parameterless one. RepeatableRead does indeed seem a more reasonable option, at least the way our system works.
Thanks!
Dawid Ciecierski
>>>>> To view this discussion on the web visit
>>>>> https://groups.google.com/d/msg/masstransit-discuss/-/_wourkOB-NIJ.
>>>>> For more options, visit https://groups.google.com/groups/opt_out.
>>>>>
>>>>>
>>>
>>> --
>>> You received this message because you are subscribed to the Google Groups
>>> "masstransit-discuss" group.
>>> To post to this group, send email to masstrans...@googlegroups.com.
>>> To unsubscribe from this group, send email to
>>> To view this discussion on the web visit
>>> https://groups.google.com/d/msg/masstransit-discuss/-/xr553B3U23cJ.
>>> For more options, visit https://groups.google.com/groups/opt_out.
>>>
>>>
>
> --
> You received this message because you are subscribed to the Google Groups
> "masstransit-discuss" group.
> To post to this group, send email to masstrans...@googlegroups.com.
> To unsubscribe from this group, send email to
> To view this discussion on the web visit
> https://groups.google.com/d/msg/masstransit-discuss/-/ytf67AXUmukJ.
>
> For more options, visit https://groups.google.com/groups/opt_out.
>
>
--
You received this message because you are subscribed to the Google Groups "masstransit-discuss" group.
To post to this group, send email to masstrans...@googlegroups.com.
To unsubscribe from this group, send email to masstransit-discuss+unsub...@googlegroups.com.
Reply all
Reply to author
Forward
0 new messages