"Collection was modified; enumeration operation may not execute" exception while RequestResponseScope.Send().

[Nik Govorov]

Hi. I have code that looks like:

ServiceBus.MakeRequest(x => x.Publish(queryMessage,
y =>
y.SendResponseTo(ServiceBus)))
.When<MetaDataReplyMessage>()
.RelatedTo(id)
.IsReceived(message => { metaData =
message.Salt; })
.TimeoutAfter(((int)ProcessingTimeOut).Seconds())
.OnTimeout(() => {throw new
MessageProcessingTimeOutException(queryMessage.ToString());})
.Send();

sometimes InvalidOperationException occured:

System.InvalidOperationException: Collection was modified; enumeration
operation may not execute.
at
System.ThrowHelper.ThrowInvalidOperationException(ExceptionResource
resource)
at
System.Collections.Generic.Dictionary`2.ValueCollection.Enumerator.MoveNext()
at
MassTransit.Pipeline.Sinks.MessageRouterBase`2.<>c__DisplayClass2.<Inspect>b__0()
at Magnum.Reflection.ReflectiveVisitorBase`1.Visit(Object obj,
Func`1 action)
at
MassTransit.Pipeline.Inspectors.PipelineInspectorBase`1.Inspect(Object
sink, Func`1 inspectChildSinks)
at
MassTransit.Pipeline.Sinks.MessageRouterBase`2.Inspect(IPipelineInspector
inspector)
at
MassTransit.Pipeline.Sinks.MessageRouter`1.<>c__DisplayClasse.<Inspect>b__c()
at Magnum.Reflection.ReflectiveVisitorBase`1.Visit(Object obj,
Func`1 action)
at
MassTransit.Pipeline.Inspectors.PipelineInspectorBase`1.Inspect(Object
sink, Func`1 inspectChildSinks)
at
MassTransit.Pipeline.Sinks.MessageRouter`1.Inspect(IPipelineInspector
inspector)
at
MassTransit.Pipeline.Sinks.MessageTranslator`2.Inspect(IPipelineInspector
inspector)
at
MassTransit.Pipeline.Sinks.MessageRouter`1.<>c__DisplayClasse.<Inspect>b__c()
at Magnum.Reflection.ReflectiveVisitorBase`1.Visit(Object obj,
Func`1 action)
at
MassTransit.Pipeline.Inspectors.PipelineInspectorBase`1.Inspect(Object
sink, Func`1 inspectChildSinks)
at
MassTransit.Pipeline.Sinks.MessageRouter`1.Inspect(IPipelineInspector
inspector)
at
MassTransit.Pipeline.MessagePipeline.<>c__DisplayClass7.<Inspect>b__6()
at Magnum.Reflection.ReflectiveVisitorBase`1.Visit(Object obj,
Func`1 action)
at
MassTransit.Pipeline.Inspectors.PipelineInspectorBase`1.Inspect(Object
sink, Func`1 inspectChildSinks)
at MassTransit.Pipeline.MessagePipeline.Inspect(IPipelineInspector
inspector)
at
MassTransit.Pipeline.Configuration.MessageRouterConfigurator.FindOrCreate[TMessage]
()
at
MassTransit.Pipeline.Configuration.CorrelatedMessageRouterConfigurator.FindOrCreate[TMessage,TKey]
()
at
MassTransit.Pipeline.MessagePipelineExtensions.Subscribe[TMessage,TKey]
(IMessagePipeline pipeline, TKey correlationId, IEndpoint endpoint)
at
MassTransit.Internal.InboundOutboundSubscriptionBinder.SubscribedTo[T,K]
(K correlationId)
at
MassTransit.Pipeline.Configuration.MessagePipelineConfigurator.<>c__DisplayClass16`2.<SubscribedTo>b__15(ISubscriptionEvent
x)
at System.Collections.Generic.List`1.ForEach(Action`1 action)
at
MassTransit.Util.RegistrationList`1.<>c__DisplayClass8.<Each>b__7(List`1
x)
at
Magnum.Threading.ReaderWriterLockedObject`1.<>c__DisplayClass1.<ReadLock>b__0(ReaderWriterLockContext
x)
at Magnum.Threading.ReaderWriterLockContext.ReadLock(Action`1
action)
at Magnum.Threading.ReaderWriterLockedObject`1.ReadLock(Action`1
action)
at MassTransit.Util.RegistrationList`1.Each(Action`1 action)
at
MassTransit.Pipeline.Configuration.MessagePipelineConfigurator.SubscribedTo[T,K]
(K correlationId)
at
MassTransit.Pipeline.Configuration.SubscriberContext.SubscribedTo[T,K]
(K correlationId)
at
MassTransit.Pipeline.Configuration.Subscribers.ConsumesForSubscriber.Connect[TMessage,TKey]
(ISubscriberContext context, For`1 consumer)
at lambda_method(ExecutionScope , ConsumesForSubscriber ,
ISubscriberContext , Object )
at
MassTransit.Pipeline.Configuration.Subscribers.ConsumesForSubscriber.<>c__DisplayClass14`1.<>c__DisplayClass18.<GetInvokerForInstance>b__10(ConsumesForSubscriber
interceptor, ISubscriberContext context, Object obj)
at
MassTransit.Pipeline.Configuration.Subscribers.ConsumesForSubscriber.<Subscribe>d__7`1.MoveNext()
at
MassTransit.Pipeline.Configuration.MessagePipelineConfigurator.<>c__DisplayClasse.<Subscribe>b__d(IPipelineSubscriber
interceptor)
at System.Collections.Generic.List`1.ForEach(Action`1 action)
at
MassTransit.Util.RegistrationList`1.<>c__DisplayClass8.<Each>b__7(List`1
x)
at
Magnum.Threading.ReaderWriterLockedObject`1.<>c__DisplayClass1.<ReadLock>b__0(ReaderWriterLockContext
x)
at Magnum.Threading.ReaderWriterLockContext.ReadLock(Action`1
action)
at Magnum.Threading.ReaderWriterLockedObject`1.ReadLock(Action`1
action)
at MassTransit.Util.RegistrationList`1.Each(Action`1 action)
at
MassTransit.Pipeline.Configuration.MessagePipelineConfigurator.Subscribe(Func`3
subscriber)
at
MassTransit.Pipeline.Configuration.MessagePipelineConfigurator.Subscribe[TComponent]
(TComponent instance)
at
MassTransit.Pipeline.MessagePipelineExtensions.<>c__DisplayClass6`1.<Subscribe>b__5(IPipelineConfigurator
x)
at MassTransit.Pipeline.MessagePipeline.Configure[V](Func`2
configurePipeline)
at
MassTransit.Pipeline.MessagePipelineExtensions.Subscribe[TComponent]
(IMessagePipeline pipeline, TComponent instance)
at MassTransit.ServiceBus.Subscribe[T](T consumer)
at
MassTransit.Internal.RequestResponse.CorrelatedResponseAction`2.SubscribeTo(IServiceBus
bus)
at
MassTransit.Internal.RequestResponse.RequestResponseScope.SubscribeToResponseMessages(UnsubscribeAction
unsubscribeToken)
at MassTransit.Internal.RequestResponse.RequestResponseScope.Send()

Thanks for help.

[Travis Smith]

How often are you seeing this happen?
Are there any specific instances of data you see that generates this exception?
Have you been able to duplicate this error in a test case?

Thanks,

-Travis

> --
> 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.
> For more options, visit this group at http://groups.google.com/group/masstransit-discuss?hl=en.
>
>

[Chris Patterson]

This is happening because you are adding a subscription to the inbound pipeline while processing a message. This isn't really a use case that I had considered since we typically use a saga to model that type of interaction to avoid nesting consumers and using up threads in the bus instance.

You could probably modify the pipeline to fit your needs, but you're starting to stretch the limits of how consumers are meant to be isolated, atomic operations.

That being said, the future of MassTransit (working towards 2.0 anyway) is to use a new message delivery pipeline that is immutable and more concurrent, allowing for these type of changes at runtime.

--
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.
For more options, visit this group at http://groups.google.com/group/masstransit-discuss?hl=en.

Chris Patterson

chrisfr...@gmail.com

[Nik Govorov]

How saga can model 'request-response' type of interaction? Can you
give a link to the example or unit test of doing something like this.
Thanks.

> > MassTransit.Pipeline.Configuration.CorrelatedMessageRouterConfigurator.Find OrCreate[TMessage,TKey]

> > ()
> >   at
> > MassTransit.Pipeline.MessagePipelineExtensions.Subscribe[TMessage,TKey]
> > (IMessagePipeline pipeline, TKey correlationId, IEndpoint endpoint)
> >   at
> > MassTransit.Internal.InboundOutboundSubscriptionBinder.SubscribedTo[T,K]
> > (K correlationId)
> >   at

> > MassTransit.Pipeline.Configuration.MessagePipelineConfigurator.<>c__Display Class16`2.<SubscribedTo>b__15(ISubscriptionEvent

> > x)
> >   at System.Collections.Generic.List`1.ForEach(Action`1 action)
> >   at
> > MassTransit.Util.RegistrationList`1.<>c__DisplayClass8.<Each>b__7(List`1
> > x)
> >   at
> > Magnum.Threading.ReaderWriterLockedObject`1.<>c__DisplayClass1.<ReadLock>b_ _0(ReaderWriterLockContext
> > x)
> >   at Magnum.Threading.ReaderWriterLockContext.ReadLock(Action`1
> > action)
> >   at Magnum.Threading.ReaderWriterLockedObject`1.ReadLock(Action`1
> > action)
> >   at MassTransit.Util.RegistrationList`1.Each(Action`1 action)
> >   at
> > MassTransit.Pipeline.Configuration.MessagePipelineConfigurator.SubscribedTo [T,K]
> > (K correlationId)
> >   at
> > MassTransit.Pipeline.Configuration.SubscriberContext.SubscribedTo[T,K]
> > (K correlationId)
> >   at

> > MassTransit.Pipeline.Configuration.Subscribers.ConsumesForSubscriber.Connec t[TMessage,TKey]

> > (ISubscriberContext context, For`1 consumer)
> >   at lambda_method(ExecutionScope , ConsumesForSubscriber ,
> > ISubscriberContext , Object )
> >   at
> > MassTransit.Pipeline.Configuration.Subscribers.ConsumesForSubscriber.<>c__D isplayClass14`1.<>c__DisplayClass18.<GetInvokerForInstance>b__10(ConsumesFo rSubscriber
> > interceptor, ISubscriberContext context, Object obj)
> >   at
> > MassTransit.Pipeline.Configuration.Subscribers.ConsumesForSubscriber.<Subsc ribe>d__7`1.MoveNext()
> >   at

> > MassTransit.Pipeline.Configuration.MessagePipelineConfigurator.<>c__Display Classe.<Subscribe>b__d(IPipelineSubscriber

> > interceptor)
> >   at System.Collections.Generic.List`1.ForEach(Action`1 action)
> >   at
> > MassTransit.Util.RegistrationList`1.<>c__DisplayClass8.<Each>b__7(List`1
> > x)
> >   at
> > Magnum.Threading.ReaderWriterLockedObject`1.<>c__DisplayClass1.<ReadLock>b_ _0(ReaderWriterLockContext

> > x)
> >   at Magnum.Threading.ReaderWriterLockContext.ReadLock(Action`1
> > action)
> >   at Magnum.Threading.ReaderWriterLockedObject`1.ReadLock(Action`1
> > action)
> >   at MassTransit.Util.RegistrationList`1.Each(Action`1 action)
> >   at

> > MassTransit.Pipeline.Configuration.MessagePipelineConfigurator.Subscribe(Fu nc`3
> > subscriber)
> >   at
> > MassTransit.Pipeline.Configuration.MessagePipelineConfigurator.Subscribe[TC omponent]
> > (TComponent instance)
> >   at
> > MassTransit.Pipeline.MessagePipelineExtensions.<>c__DisplayClass6`1.<Subscr ibe>b__5(IPipelineConfigurator

> > x)
> >   at MassTransit.Pipeline.MessagePipeline.Configure[V](Func`2
> > configurePipeline)
> >   at
> > MassTransit.Pipeline.MessagePipelineExtensions.Subscribe[TComponent]
> > (IMessagePipeline pipeline, TComponent instance)
> >   at MassTransit.ServiceBus.Subscribe[T](T consumer)
> >   at
> > MassTransit.Internal.RequestResponse.CorrelatedResponseAction`2.SubscribeTo (IServiceBus
> > bus)
> >   at
> > MassTransit.Internal.RequestResponse.RequestResponseScope.SubscribeToRespon seMessages(UnsubscribeAction
> > unsubscribeToken)
> >   at MassTransit.Internal.RequestResponse.RequestResponseScope.Send()
>
> > Thanks for help.
>
> > --
> > 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.
> > For more options, visit this group athttp://groups.google.com/group/masstransit-discuss?hl=en.
>
> Chris Patterson

> chrisfromtu...@gmail.com