Is it possible to specify individual transactional queues?

[Mike Hadlow]

We can specify transactional queues like this:

MsmqEndpointConfigurator.Defaults(x =>
{
x.CreateTransactionalQueues = true;
});

But this is saying that every queue my service contacts should be
transactional. The main piece of MT infrastructure, RuntimeServices,
creates non-transactional queues. If I configure my service with
transactional queues as above, I get exceptions saying that it expects
the mt_subscriptions queue to be transactional when it's not.

So my question is: Is it possible to configure my service with a
transactional main queue, but a non-transactional control queue? Is
this a good idea?

Thanks
Mike

[Chris Patterson]

Interesting -- if the queue is not being created and isn't the primary queue of the service bus (or a control bus) I'm not sure why it is upset that the queue is not transactional.

Chris Patterson

chrisfr...@gmail.com

[Mike Hadlow]

Thanks Chris,

Here's the stack trace (sorry, I should have included this earlier):

Unhandled Exception: Magnum.StateMachine.StateMachineException:
Exception occurred in Topshelf.Internal.ServiceController`1
[[TPR.Bus.Common.BusSubscriber, TPR.Bus.Common,Version=1.0.0.0,
Culture=neutral, PublicKeyToken=null]] during state
Initial while handling OnStart --->
MassTransit.Exceptions.EndpointException: msmq://tpr-itss-012/mt_subscriptions
=> Failed to create MSMQ endpoint --->
MassTransit.Exceptions.TransportException: msmq://tpr-itss-012/mt_subscriptions
=> Failed to create MSMQ transport --->
MassTransit.Exceptions.TransportException: msmq://tpr-itss-012/mt_subscriptions
=> There was a problem managing the transport --->
MassTransit.Exceptions.TransportException: msmq://tpr-itss-012/mt_subscriptions
=> The transport is non-transactional but a transactional transport
was requested
at
MassTransit.Transports.Msmq.MsmqTransportFactory.<>c__DisplayClass1.<ValidateLocalTransport>b__0
(IEndpointManagement q) in d:\Source\masstransit\src\Transports
\MassTransit.Transports.Msmq\MsmqTransportFactory.cs:line 68
at MassTransit.Transports.Msmq.MsmqEndpointManagement.Manage
(IMsmqEndpointAddress address, Action`1 action) in d:\Source
\masstransit\src\Transports\MassTransit.Transports.Msmq
\MsmqEndpointManagement.cs:line 138
--- End of inner exception stack trace ---
at MassTransit.Transports.Msmq.MsmqEndpointManagement.Manage
(IMsmqEndpointAddress address, Action`1 action) in d:\Source
\masstransit\src\Transports\MassTransit.Transports.Msmq
\MsmqEndpointManagement.cs:line 142
at
MassTransit.Transports.Msmq.MsmqTransportFactory.ValidateLocalTransport
(CreateMsmqTransportSettings settings) in d:\Source\masstransit\src
\Transports\MassTransit.Transports.Msmq\MsmqTransportFactory.cs:line
52
at
MassTransit.Transports.Msmq.MsmqTransportFactory.NewLocalTransport
(CreateMsmqTransportSettings settings) in d:\Source\masstransit\src
\Transports\MassTransit.Transports.Msmq\MsmqTransportFactory.cs:line
37
at MassTransit.Transports.Msmq.MsmqTransportFactory.New
(CreateMsmqTransportSettings settings) in d:\Source\masstransit\src
\Transports\MassTransit.Transports.Msmq\MsmqTransportFactory.cs:line
24
--- End of inner exception stack trace ---
at MassTransit.Transports.Msmq.MsmqTransportFactory.New
(CreateMsmqTransportSettings settings) in d:\Source\masstransit\src
\Transports\MassTransit.Transports.Msmq\MsmqTransportFactory.cs:line
31
at MassTransit.Transports.Msmq.MsmqEndpointFactory.New
(CreateMsmqEndpointSettings settings) in d:\Source\masstransit\src
\Transports\MassTransit.Transports.Msmq\MsmqEndpointFactory.cs:line 38
--- End of inner exception stack trace ---
at MassTransit.Transports.Msmq.MsmqEndpointFactory.New
(CreateMsmqEndpointSettings settings) in d:\Source\masstransit\src
\Transports\MassTransit.Transports.Msmq\MsmqEndpointFactory.cs:line 51
at MassTransit.Transports.Msmq.MsmqEndpointConfigurator.Create() in
d:\Source\masstransit\src\Transports\MassTransit.Transports.Msmq
\MsmqEndpointConfigurator.cs:line 29
at MassTransit.Transports.Msmq.MsmqEndpointConfigurator.New
(Action`1 action) in d:\Source\masstransit\src\Transports
\MassTransit.Transports.Msmq\MsmqEndpointConfigurator.cs:line 46
at MassTransit.Transports.Msmq.MsmqEndpoint.ConfigureEndpoint(Uri
uri, Action`1 configurator) in d:\Source\masstransit\src\Transports
\MassTransit.Transports.Msmq\MsmqEndpoint.cs:line 206
at lambda_method(ExecutionScope , Uri , Action`1 )
at MassTransit.Internal.EndpointFactory.BuildEndpoint(Uri uri) in d:
\Source\masstransit\src\MassTransit\Internal\EndpointFactory.cs:line
123
at
MassTransit.Internal.EndpointFactory.<>c__DisplayClass2.<GetEndpoint>b__0
() in d:\Source\masstransit\src\MassTransit\Internal
\EndpointFactory.cs:line 86
at
Magnum.Threading.ReaderWriterLockedDictionary`2.<>c__DisplayClassb.<Retrieve>b__a
(Dictionary`2 x) in d:\Home\Magnum\trunk\src\Magnum\Threading
\ReaderWriterLockedDictionary.cs:line 82
at
Magnum.Threading.ReaderWriterLockedObject`1.<>c__DisplayClass19`1.<WriteLock>b__18
(ReaderWriterLockContext x) in d:\Home\Magnum\trunk\src\Magnum
\Threading\ReaderWriterLockedObject.cs:line 112
at Magnum.Threading.ReaderWriterLockContext.WriteLock[V](Func`2
action) in d:\Home\Magnum\trunk\src\Magnum\Threading
\ReaderWriterLockContext.cs:line 165
at Magnum.Threading.ReaderWriterLockedObject`1.WriteLock[V](Func`2
action) in d:\Home\Magnum\trunk\src\Magnum\Threading
\ReaderWriterLockedObject.cs:line 112
at Magnum.Threading.ReaderWriterLockedDictionary`2.Retrieve(TKey
key, Func`1 valueProvider) in d:\Home\Magnum\trunk\src\Magnum\Threading
\ReaderWriterLockedDictionary.cs:line 77
at MassTransit.Internal.EndpointFactory.GetEndpoint(Uri uri) in d:
\Source\masstransit\src\MassTransit\Internal\EndpointFactory.cs:line
88
at
MassTransit.Services.Subscriptions.Client.SubscriptionClient.Start
(IServiceBus bus) in d:\Source\masstransit\src\MassTransit\Services
\Subscriptions\Client\SubscriptionClient.cs:line 115
at MassTransit.Internal.ServiceContainer.Start() in d:\Source
\masstransit\src\MassTransit\Internal\ServiceContainer.cs:line 48
at MassTransit.ServiceBus.Start() in d:\Source\masstransit\src
\MassTransit\ServiceBus.cs:line 284
at MassTransit.Configuration.ServiceBusConfigurator.Create() in d:
\Source\masstransit\src\MassTransit\Configuration
\ServiceBusConfigurator.cs:line 130
at MassTransit.Configuration.ServiceBusConfigurator.New(Action`1
action) in d:\Source\masstransit\src\MassTransit\Configuration
\ServiceBusConfigurator.cs:line 195
at
MassTransit.WindsorIntegration.MassTransitFacility.RegisterServiceBus
(String id, String endpointUri, Action`1 configAction) in d:\Source
\masstransit\src\Containers\MassTransit.WindsorIntegration
\MassTransitFacility.cs:line 102
at
MassTransit.WindsorIntegration.MassTransitFacility.ConfigureServiceBus
() in d:\Source\masstransit\src\Containers
\MassTransit.WindsorIntegration\MassTransitFacility.cs:line 74
at MassTransit.WindsorIntegration.MassTransitFacility.Init() in d:
\Source\masstransit\src\Containers\MassTransit.WindsorIntegration
\MassTransitFacility.cs:line 39
at Castle.MicroKernel.Facilities.AbstractFacility.Init(IKernel
kernel, IConfiguration facilityConfig)
at Castle.MicroKernel.DefaultKernel.AddFacility(String key,
IFacility facility)
at Castle.Windsor.WindsorContainer.AddFacility(String key,
IFacility facility)
at
TPR.Bus.Common.ServiceConfigurationBuilder.<>c__DisplayClass11.<CreateContainer>b__f
() in D:\Projects\Releases\Release_11\Trunk\Assemblies\TPR.Bus.Common
\ServiceConfigurationBuilder.cs:line 179
at
TPR.Bus.Common.ServiceConfigurationBuilder.<>c__DisplayClassc.<CreateServiceLocator>b__a
() in D:\Projects\Releases\Release_11\Trunk\Assemblies\TPR.Bus.Common
\ServiceConfigurationBuilder.cs:line 149
at Topshelf.Internal.ServiceController`1.get_ServiceLocator()
at Topshelf.Internal.ServiceController`1.BuildInstance()
at Topshelf.Internal.ServiceController`1.<.cctor>b__2
(ServiceController`1 sc)
at Magnum.StateMachine.LambdaAction`1.Execute(T instance, Event
event, Object parameter) in d:\Home\Magnum\trunk\src\Magnum
\StateMachine\LambdaAction.cs:line 30
at Magnum.StateMachine.EventActionList`1.Execute(T stateMachine,
Event event, Object parameter) in d:\Home\Magnum\trunk\src\Magnum
\StateMachine\EventActionList.cs:line 99
--- End of inner exception stack trace ---
at Magnum.StateMachine.ExceptionActionDictionary`1.HandleException
(T stateMachine, Event event, Object parameter, Exception exception)
in d:\Home\Magnum\trunk\src\Magnum\StateMachine
\ExceptionActionDictionary.cs:line 60
at Magnum.StateMachine.EventActionList`1.Execute(T stateMachine,
Event event, Object parameter) in d:\Home\Magnum\trunk\src\Magnum
\StateMachine\EventActionList.cs:line 107
at Magnum.StateMachine.EventActionBase`1.Execute(T instance, Event
event, Object parameter) in d:\Home\Magnum\trunk\src\Magnum
\StateMachine\EventActionBase.cs:line 39
at Magnum.StateMachine.State`1.RaiseEvent(T instance, BasicEvent`1
eevent, Object value) in d:\Home\Magnum\trunk\src\Magnum\StateMachine
\State.cs:line 75
at Magnum.StateMachine.StateMachine`1.RaiseEvent(Event raised) in d:
\Home\Magnum\trunk\src\Magnum\StateMachine\StateMachine.cs:line 99
at Topshelf.Internal.ServiceController`1.Start()
at Topshelf.Internal.ServiceCoordinator.Start()
at Topshelf.Internal.Hosts.ConsoleHost.Run()
at Topshelf.Internal.Actions.RunAsConsoleAction.Do
(IRunConfiguration configuration)
at Topshelf.Runner.Host(IRunConfiguration configuration, String[]
args)
at TPR.Bus.NotifiableEvents.Program.Main(String[] args) in D:
\Projects\Releases\Release_11\Trunk\NtService\TPR.Bus.No
tifiableEvents\Program.cs:line 26
Press any key to continue . . .

> chrisfromtu...@gmail.com

[Mike Hadlow]

The problem seems to be in MsmqTransportFactory.ValidateLocalTransport
method. This checks that any existing local queues have the same
transactional setting as CreateMsmqTransportSettings. Since
CreateMsmqTransportSettings is set with MsmqEndpointConfigurator for
all local queues, you can't have a mix of transactional and non-
transactional local queues. If you are running RuntimeServices locally
(which uses non-transactional queues) the exception will be thrown if
your service config has:

MsmqEndpointConfigurator.Defaults(config =>
{
config.CreateTransactionalQueues = true;
});

This seems to fix it
(MassTransit.Transports.Msmq.MsmqTransportFactory):

private static void ValidateLocalTransport(CreateMsmqTransportSettings
settings)
{
MsmqEndpointManagement.Manage(settings.Address, q =>
{
if (!q.Exists)
{
if (!settings.CreateIfMissing)
throw new TransportException(settings.Address.Uri,
"The transport does not exist and automatic
creation is not enabled");

q.Create(settings.Transactional);
}
// added this else clause
else
{
settings.Transactional = q.IsTransactional;
}

// removed this validation check
// if (q.IsTransactional && !settings.Transactional)
// throw new TransportException(settings.Address.Uri,
// "The transport is transactional but a non-
transactional transport was requested");
//
// if (!q.IsTransactional && settings.Transactional)
// throw new TransportException(settings.Address.Uri,
// "The transport is non-transactional but a
transactional transport was requested");
});
}

Mike

On Oct 31, 2:09 am, Chris Patterson <chrisfromtu...@gmail.com> wrote:

> chrisfromtu...@gmail.com

[Chris Patterson]

Cool, I will put this in today.

I'm going to add another setting, "RequireTransactionalQueues" that will check the queue type and only call this code if it is set (which in this case, will default to false).

Thanks Man!

Chris

PS: Did I reply about message expiration yet?

_endpoint.Send(message, x => x.ExpiresAt(5.Minutes().FromNow()));

Chris Patterson

chrisfr...@gmail.com

[Chris Patterson]

Fix for this is committed.

[Mike Hadlow]

Excellent, thanks Chris!

Yes, I saw about the message expiration. It would be very nice if this
was added to all infrastructure stuff (hearbeat, timeout etc) ;)

Cheers
Mike

On Nov 2, 2:47 pm, Chris Patterson <chrisfromtu...@gmail.com> wrote:
> Fix for this is committed.
>

[Chris Patterson]

We are going to add it to heartbeats for the health service, not sure about all the others yet. Timeouts have a solid business value regardless of timeout.

[Mike Hadlow]

Sorry dude, it's still not quite right. When the queue doesn't already
exist, you are asking it if it's transactional before creating it
which throws. Here's my modified ValidateLocalTransport (in
MsmqTransportFactory):

private static LocalTransportSettings ValidateLocalTransport
(CreateMsmqTransportSettings settings)
{
var result = new LocalTransportSettings
{
Transactional = false,
};

MsmqEndpointManagement.Manage(settings.Address, q =>
{
// throws if q doesn't exist yet.
// result.Transactional = q.IsTransactional;

if (!q.Exists)
{
if (!settings.CreateIfMissing)
throw new TransportException(settings.Address.Uri,
"The transport does not exist and automatic
creation is not enabled");

q.Create(settings.Transactional);
}

if (settings.RequireTransactional)
{
if (!q.IsTransactional && settings.Transactional)
throw new TransportException(settings.Address.Uri,

"The transport is non-transactional but a
transactional transport was requested");
}

// moved result settting to after q is created
result.Transactional = q.IsTransactional;
});

return result;
}

Yeah, yeah, I know I should really submit a patch :p

Thanks
Mike

On Nov 2, 9:21 pm, Chris Patterson <chrisfromtu...@gmail.com> wrote:
> We are going to add it to heartbeats for the health service, not sure about
> all the others yet. Timeouts have a solid business value regardless of
> timeout.
>

[Chris Patterson]

Geez I'm a klutz. We're getting a lot of mileage out of this with Travis doing some work and we'll have a patch in by the end of the day that handles a lot of this much more nicely.