nimbusapi / nimbus Goto Github PK
View Code? Open in Web Editor NEWNimbus is a .NET client library to provide an easy abstraction over common messaging frameworks.
Home Page: http://nimbusapi.com/
License: MIT License
Nimbus is a .NET client library to provide an easy abstraction over common messaging frameworks.
Home Page: http://nimbusapi.com/
License: MIT License
At the moment if you publish an event that hasn't been registered, there is no topic for it and the service bus throws a FaultException: 40400: Endpoint not found..
We could look up known types on every publish and throw a more helpful BusException with a useful error message.
Discuss
The default parts separator for creating the Path to service bus entities in Nimbus is '.' (a period) which is fine in most situations though it can be problemmatic in URLs.
A specific example is that by default IIS will treat a request ending with something that looks like a file as a request for a static resource and returns 404 when it can't find the file on the file system. You can work around this by adding a trailing '/' to the URL or by specifically adding an exception in web.config.
I'm just wondering if period is the most appropriate parts separator?
Perhaps Nimbus could offer some customisable conventions?
The SBWS/ASB MessagePump only allows concurrent handling of messages AFTER .Complete (and friends) is called. Until the message is 'complete' it will only call 1 handler at a time. This can be worked around by using a custom message pump instead of the in-built one. This issue is not obvious when processing commands quickly (more-or-less in-proc), but if a handler has to make an external api/resource call, it hold up many messages.
You can see an example of this in the following gist:
https://gist.github.com/lukeschafer/9797555
Obviously sub in your queue name and connection string. Toggle 'useCustomMessagePump' to see the difference.
7e90b59 introduced a change to subscription naming and we happen to have a lot of Request message names that cause the Subscription Path to break the 50 character limit imposed by Service Bus.
One example is: nextgen.appserver.getrecordsmodifiedsincerequesthandler
public static string SubscriptionNameFor(string applicationName, string instanceName, Type handlerType)
{
return Sanitize(string.Join(".", new[] {applicationName, instanceName, handlerType.Name}));
}
The offending code in the Microsoft.ServiceBus assembly in NamespaceManager (v2.1.3.0):
NamespaceManager.CheckValidEntityName(description.TopicPath, 260, "description.TopicPath");
NamespaceManager.CheckValidEntityName(description.Name, 50, false, "description.Name");
This issue was raised here also Particular/NServiceBus.Azure#19
It would be nice if, when using the serilog sink, it destructured things like the message objects. I had a bit of a play with trying it in the SerilogLogger or within Serilog, however it didn't seem right.
I best solution I can think of is some way of specifying a different message format for serilog than for the console logger:
No, not via magic pixie dust or the NSA's QUANTUM gadgetry; merely that we should at least allow applications to launch even when they can't connect to a bus endpoint.
It might be nice to allow commands and events to be queued for later transmission but that's a separate issue - I don't really have an opinion yet as to whether that's a responsibility for the bus or for the app. Either way, it's not what this issue is about.
OK folks
One of my little holiday tasks is going to be putting some documentation together about the concepts and usage of Nimbus.
Open to any ideas or suggestions as to what this might look like. What would you like to see ? Or if there examples of projects "doing it right" that would be cool too.
For those who need it. (Or are convinced they do)
Option to use different object serialisers for BrokeredMessage, which is currently the binary XML serialiser.
Currently when a new instance of BrokeredMessage is created, the binary XML serialiser is used (default option). I would like the option to specify a custom serialiser (that implements XmlObjectSerializer) when I build the BusBuilderConfiguration in my application setup.
This option would be useful when using non .NET clients and when using Service Bus Explorer to view messages.
I've listed two possible implementations below that could be used.
Example 1
Implementation
BusBuilderConfiguration WithSerializer<T>(this BusBuilderConfiguration configuration) where T : XmlObjectSerializer;
Usage
.WithSerializer<DataContractSerializer>()
Example 2
Implementation
BusBuilderConfiguration WithSerializer(this BusBuilderConfiguration configuration, Type serializerType)
Usage
.WithSerializer(typeof(DataContractSerializer))
We can have
public class Foo: IHandleCommand<SomeCommand> {...}
registered from one assembly but the assembly containing SomeCommand doesn't currently need to be registered. This isn't going to end well.
When a dispatch failure occurs the Exception Type, Message and StackTrace are recorded as message properties. Add timestamp, failed path, executing identity and machine name to help diagnose dispatch failures.
We're currently not executing the Windsor integration test assembly because we don't have a wildcard match on test assemblies.
If we rename our Nimbus.Tests assembly to Nimbus.UnitTests then we can use */UnitTest.dll and */IntegrationTest.dll as our test assembly wildcards in each separate build step.
The serializer used by Azure Message Bus can't handle things like protected properties, classes without default constructors etc.
On startup we should scan for any types that we're supposed to care about and throw an exception if they're going to cause a serialization failure.
I setup a very basic script in LINQPad to try Nimbus out, but got the following error response from Azure Service Bus when I tried to run it:
The remote server returned an error: (400) Bad Request. SubCode=40000. 'https://myappname.servicebus.windows.net/Q.UserQuery+TestCommand/?api-version=2013-10' contains character(s) that is not allowed by Service Bus. Entity segments can contain only letters, numbers, periods (.), hyphens (-), and underscores (_)..TrackingId:60e8d3be-f322-4da9-adb3-ebc120d858b9_G8,TimeStamp:12/11/2013 12:44:48 AM
I don't really know much of how .NET names types internally, but this page seems to indicate that a "+" is used for nested classes: http://msdn.microsoft.com/en-us/library/w3f99sx1(v=vs.110).aspx
Michael
Possibly a heisenbug but I got an InvalidOperationException
in BatchingMessageSender
at the end of FlushMessages()
where it checks _outboundQueue.Any()
(here), with ye olde Collection was modified; enumeration operation may not execute.
for more information.
The Any()
call is outside of a lock
, would this the culprit?
Running up again with the same scenario worked, so I can't easily repro this.
We've noticed that after running for a period of time the MessageReceivers can end up going into a CPU-expensive loop requiring a recycle of the application to recover.
I was able to repro, by sheer accident, on my local machine and we noticed that after pausing the console app for a minute while taking the stack trace, the app seemed to recover, but I'm not quite sure what state the app was in afterwards.
Microsoft.ServiceBus.Messaging.MessagingCommunicationException: Error during communication with Service Bus. Check the connection information, then retry. ---> System.ServiceModel.CommunicationObjectFaultedException: Internal Server Error: The server did not provide a meaningful reply; this might be caused by a premature session shutdown..TrackingId:2894fb95-7070-4e81-975f-fb71e7eaa5de, Timestamp:22/04/2014 12:46:37 AM
Server stack trace:
Exception rethrown at [0]:
at Microsoft.ServiceBus.Common.AsyncResult.End[TAsyncResult](IAsyncResult result)
at Microsoft.ServiceBus.Messaging.Sbmp.DuplexRequestBindingElement.DuplexRequestSessionChannel.DuplexCorrelationAsyncResult.End(IAsyncResult result)
at Microsoft.ServiceBus.Messaging.Sbmp.DuplexRequestBindingElement.DuplexRequestSessionChannel.EndRequest(IAsyncResult result)
at Microsoft.ServiceBus.Messaging.Channels.ReconnectBindingElement.ReconnectChannelFactory`1.RequestSessionChannel.RequestAsyncResult.<GetAsyncSteps>b__4(RequestAsyncResult thisPtr, IAsyncResult r)
at Microsoft.ServiceBus.Messaging.IteratorAsyncResult`1.StepCallback(IAsyncResult result)
Exception rethrown at [1]:
at Microsoft.ServiceBus.Common.AsyncResult.End[TAsyncResult](IAsyncResult result)
at Microsoft.ServiceBus.Common.AsyncResult`1.End(IAsyncResult asyncResult)
at Microsoft.ServiceBus.Messaging.Channels.ReconnectBindingElement.ReconnectChannelFactory`1.RequestSessionChannel.EndRequest(IAsyncResult result)
at Microsoft.ServiceBus.Messaging.Sbmp.RedirectBindingElement.RedirectContainerChannelFactory`1.RedirectContainerSessionChannel.RequestAsyncResult.<>c__DisplayClass17.<GetAsyncSteps>b__a(RequestAsyncResult thisPtr, IAsyncResult r)
at Microsoft.ServiceBus.Messaging.IteratorAsyncResult`1.StepCallback(IAsyncResult result)
Exception rethrown at [2]:
at Microsoft.ServiceBus.Common.AsyncResult.End[TAsyncResult](IAsyncResult result)
at Microsoft.ServiceBus.Common.AsyncResult`1.End(IAsyncResult asyncResult)
at Microsoft.ServiceBus.Messaging.Sbmp.RedirectBindingElement.RedirectContainerChannelFactory`1.RedirectContainerSessionChannel.EndRequest(IAsyncResult result)
at Microsoft.ServiceBus.Messaging.Channels.ReconnectBindingElement.ReconnectChannelFactory`1.RequestSessionChannel.RequestAsyncResult.<GetAsyncSteps>b__4(RequestAsyncResult thisPtr, IAsyncResult r)
at Microsoft.ServiceBus.Messaging.IteratorAsyncResult`1.StepCallback(IAsyncResult result)
Exception rethrown at [3]:
at Microsoft.ServiceBus.Common.AsyncResult.End[TAsyncResult](IAsyncResult result)
at Microsoft.ServiceBus.Common.AsyncResult`1.End(IAsyncResult asyncResult)
at Microsoft.ServiceBus.Messaging.Channels.ReconnectBindingElement.ReconnectChannelFactory`1.RequestSessionChannel.EndRequest(IAsyncResult result)
at Microsoft.ServiceBus.Messaging.Sbmp.SbmpMessageReceiver.EndReceiveCommand(IAsyncResult result, IEnumerable`1& messages)
--- End of inner exception stack trace ---
at Microsoft.ServiceBus.Common.AsyncResult.End[TAsyncResult](IAsyncResult result)
at Microsoft.ServiceBus.Common.AsyncResult`1.End(IAsyncResult asyncResult)
at Microsoft.ServiceBus.Messaging.MessageReceiver.RetryReceiveAsyncResult.TryReceiveEnd(IAsyncResult r, IEnumerable`1& messages)
at Microsoft.ServiceBus.Messaging.MessageReceiver.EndTryReceive(IAsyncResult result, IEnumerable`1& messages)
at Microsoft.ServiceBus.Messaging.MessageReceiver.EndReceiveBatch(IAsyncResult result)
at Microsoft.ServiceBus.Messaging.SubscriptionClient.EndReceiveBatch(IAsyncResult result)
at System.Threading.Tasks.TaskFactory`1.FromAsyncCoreLogic(IAsyncResult iar, Func`2 endFunction, Action`1 endAction, Task`1 promise, Boolean requiresSynchronization)
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at System.Runtime.CompilerServices.TaskAwaiter`1.GetResult()
at Nimbus.Infrastructure.MessageSendersAndReceivers.NimbusSubscriptionMessageReceiver.<FetchBatch>d__3.MoveNext() in c:\\projects\\nextgenrelational\\Nimbus\\src\\Nimbus\\Infrastructure\\MessageSendersAndReceivers\\NimbusSubscriptionMessageReceiver.cs:line 42<---
We're going to need this for auditing, at least, but more likely we'll want it for an "Ops Dash" style feature before that.
I'm thinking just something along the lines of:
public class TheAuditors: IHandleMulticastEvent<AuditEvent>
{
...
}
where the AuditEvent looks something like this:
public class AuditEvent: IBusEvent
{
public BrokeredMessage AuditedMessage { get; set; }
}
Thoughts?
Does Nimbus support that? I've had to put lots ot DataContract/DataMember and KnownType all over my messages to get it working...
Can I swap the default serializer?
I mean things like:
public class TimelineNotification : IBusCommand
{
public ITimelineEventData Data { get; set; }
}
The fact these methods are Task-returning makes it fairly obvious, just a thought.
I personally forgot to await a call to Bus.Send() and didn't catch the exception that occurred due to an assembly registration error... Seeing something like SendAsync without an await might provide an early warning sign that something's amiss.
One friction point encountered with frameworks that abstract registration into the container, like Nimbus does right now, is that this often makes it really hard to deal with some more advanced scenarios like customise how a particular handler is registered, or test the registration (a convention test).
Are there any plans to explore alternative ways of integrating with a container, to make it more flexible?
The 2.2.x.x version of WindowsAzure.ServiceBus sends an API version string of 2014-01, which isn't supported by Windows Service Bus 1.1. This means that we can't currently point Nimbus at on-premises installations. WSB1.1 only supports an API version of 2013-07.
It would be possible to have a naming collision with queues.
Azure queues are lowercased, but it would be possible to have two message types with the same name and different casing. An edge case, but something we should test for at scan time.
Perhaps a dictionary of known queue names and if we try to spin it up twice we go bang .
Rather than everyone rolling their own dead-letter support, I believe it makes sense to support several dead-letter features as first-class citizens. Some examples would be:
Our sample code currently looks like this for non-container installations:
var typeProvider = new AssemblyScanningTypeProvider(Assembly.GetExecutingAssembly());
var messageBroker = new DefaultMessageBroker(typeProvider);
var bus = new BusBuilder().Configure()
.WithNames("MyTestSuite", Environment.MachineName)
.WithConnectionString(CommonResources.ConnectionString)
.WithTypesFrom(typeProvider)
.WithCommandBroker(messageBroker)
.WithRequestBroker(messageBroker)
.WithMulticastEventBroker(messageBroker)
.WithCompetingEventBroker(messageBroker)
.WithMulticastRequestBroker(messageBroker)
.WithDefaultTimeout(TimeSpan.FromSeconds(10))
.Build();
It would be nice if it could look like this instead:
var typeProvider = new AssemblyScanningTypeProvider(Assembly.GetExecutingAssembly());
var messageBroker = new DefaultMessageBroker(typeProvider);
var bus = new BusBuilder().Configure()
.WithNames("MyTestSuite", Environment.MachineName)
.WithConnectionString(CommonResources.ConnectionString)
.WithTypesFrom(typeProvider)
.WithDefaultBroker(messageBroker)
.WithDefaultTimeout(TimeSpan.FromSeconds(10))
.Build();
You shouldn't declare finalizers unless you have to. You should only declare a finalizer if you have unmanaged resources to dispose. The CLR deals with finalizable objects differently, even if SuppressFinalize is called. More info on msdn
public class Bus : IBus, IDisposable
{
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
// No need in finalizer
~Bus()
{
Dispose(false);
}
protected virtual void Dispose(bool disposing)
{
// Because no unmanaged resources here
if (!disposing) return;
//...
}
}
I re-enabled integration tests and pushed some changes that broke the test suite but not compilation. The build (1.0.0.129) was marked as failed but there's still a package on nuget.org for that build number.
I've checked and all the TC build steps are set to only execute if the previous step was successful. Odd.
Nimbus.MessageContracts depends on Nimbus. Both are built as part of the same solution and packed via nuget.exe as part of the same build pass.
Looks like the call to nuget.exe pack doesn't set package version dependencies, so Nimbus itself only has a dependency on Nimbus.MessageContracts v0.0.1 no matter what version the Nimbus package is.
We've experienced several times where a server will go into an endless loop until it crashes with an OutOfMemoryException (about once per day over the last week). I've finally been able to reproduce the issue and I have a theory about the cause.
We've got 600MB of the following log message:
2014-05-08 07:36:59.537 +00:00 [Debug] Flushing outbound message queue "q.mymessage" (1 messages)
2014-05-08 07:36:59.537 +00:00 [Debug] Flushing outbound message queue "q.mymessage" (1 messages)
2014-05-08 07:36:59.537 +00:00 [Debug] Flushing outbound message queue "q.mymessage" (1 messages)
2014-05-08 07:36:59.537 +00:00 [Debug] Flushing outbound message queue "q.mymessage" (1 messages)
2014-05-08 07:36:59.537 +00:00 [Debug] Flushing outbound message queue "q.mymessage" (1 messages)
2014-05-08 07:36:59.537 +00:00 [Debug] Flushing outbound message queue "q.mymessage" (1 messages)
2014-05-08 07:36:59.537 +00:00 [Debug] Flushing outbound message queue "q.mymessage" (1 messages)
2014-05-08 07:36:59.537 +00:00 [Debug] Flushing outbound message queue "q.mymessage" (1 messages)
2014-05-08 07:36:59.537 +00:00 [Debug] Flushing outbound message queue "q.mymessage" (1 messages)
2014-05-08 07:36:59.537 +00:00 [Debug] Flushing outbound message queue "q.mymessage" (1 messages)
2014-05-08 07:36:59.537 +00:00 [Debug] Flushing outbound message queue "q.mymessage" (1 messages)
Looking at the code in https://github.com/NimbusAPI/Nimbus/blob/master/src/Nimbus/Infrastructure/MessageSendersAndReceivers/BatchingMessageSender.cs#L75
If the underlying MessageSender throws immediately, say it gets disposed, the BatchingMessageSender will go into an endless loop and never recover because the MessageSender isn't recycled.
A couple of thoughts:
Is filter the best term? It's synonymous with the MVC filters, so probably...
e.g.
IFilterOutgoingMessages
IFilterOutgoingMessagesOf<T>
IFilterIncomingMessages
IFilterIncomingMessagesOf<T>
Possible definitions
public interface IFilterOutgoingMessagesOf<T> {
T FilterOut(T message);
BrokeredMessage FilterOut(BrokeredMessage brokeredMessage, T originalMessage);
}
public interface IFilterIncomingMessagesOf<T> {
T FilterIn(BrokeredMessage brokeredMessage, T originalMessage);
}
Registering
Would normally just register the components in the container, but it may be useful to have something like:
new BusBuilder().Configure()
.WithOutgoingMessageFilter<DataAnnotationValidator>() // DataAnnotationValidator : IFilterOutgoingMessages
.WithOutgoingMessageFilter<DataBusMutator>() // DataBusMutator : IFilterOutgoingMessages, IFilterIncomingMessages
.WithIncomingMessageFilter<DataBusMutator>() // DataBusMutator : IFilterOutgoingMessages, IFilterIncomingMessages
.WithOutgoingMessageFilter<
The implementations can return original input (modified or whatever) or return new instances. They have the ability to throw exception and cancel the incoming/outgoing message. Would be transient per message (to enable state if we require peeking at multiple stages of the message)
Example usages:
Outgoing filter to run validation on the message using validation attributes (built-in?)
Outgoing filter to detect 'DataBus' and apply changes to the message - this might be better if it can also modify the brokeredmessage
Incoming filter to detect 'DataBus' and apply changes to the message - this might be better if it can also read the brokeredmessage
Consider logging dispatch exceptions as Warnings until the message will be dead-lettered at which point escalate it to an Error. Couple of thoughts about why:
Some handlers need to have a set maximum number of instances. We've seen this in the wild with spinning up a pool of PhantomJS instances, for instance, as well as where third-party APIs limit incoming connections.
We need a nice way to specify how many concurrently-executing instances of a handler of a particular type should permit, as well as a sensible default.
We've been experiencing an intermittent issue with shutting down a Windows Service that uses Nimbus - it can take up to 10 minutes to shutdown. I finally reproduced this locally and have some exception stack traces (see end of post).
Inspecting the call stack of the threads that are still running I can see the following lines of code are blocked waiting for other tasks to complete:
Line 121 of Bus.cs: Task.WaitAll(messagePumpStopTasks);
Line 87 of NimbusMessageReceiver in the FetchBatchInternal method: await Task.Delay(100);
After letting it run for about 30-60 seconds I started seeing these exceptions being thrown:
2014-04-15 13:15:23.694 +10:00 [Error] Worker exception in "NimbusQueueMessageReceiver" for "q.nextgen.messaging.contracts.commands.updateeventsettingsbuscommand"
System.AggregateException: One or more errors occurred. ---> Microsoft.ServiceBus.Messaging.MessagingCommunicationException: Error during communication with Service Bus. Check the connection information, then retry. ---> System.ServiceModel.CommunicationObjectFaultedException: Internal Server Error: The server did not provide a meaningful reply; this might be caused by a premature session shutdown..TrackingId:ce8706ce-c459-41a8-b6e5-5398f85055d9, Timestamp:15/04/2014 3:15:19 AM
Server stack trace:
Exception rethrown at [0]:
at Microsoft.ServiceBus.Common.AsyncResult.End[TAsyncResult](IAsyncResult result)
at Microsoft.ServiceBus.Messaging.Sbmp.DuplexRequestBindingElement.DuplexRequestSessionChannel.DuplexCorrelationAsyncResult.End(IAsyncResult result)
at Microsoft.ServiceBus.Messaging.Sbmp.DuplexRequestBindingElement.DuplexRequestSessionChannel.EndRequest(IAsyncResult result)
at Microsoft.ServiceBus.Messaging.Channels.ReconnectBindingElement.ReconnectChannelFactory`1.RequestSessionChannel.RequestAsyncResult.<GetAsyncSteps>b__4(RequestAsyncResult thisPtr, IAsyncResult r)
at Microsoft.ServiceBus.Messaging.IteratorAsyncResult`1.StepCallback(IAsyncResult result)
Exception rethrown at [1]:
at Microsoft.ServiceBus.Common.AsyncResult.End[TAsyncResult](IAsyncResult result)
at Microsoft.ServiceBus.Common.AsyncResult`1.End(IAsyncResult asyncResult)
at Microsoft.ServiceBus.Messaging.Channels.ReconnectBindingElement.ReconnectChannelFactory`1.RequestSessionChannel.EndRequest(IAsyncResult result)
at Microsoft.ServiceBus.Messaging.Sbmp.RedirectBindingElement.RedirectContainerChannelFactory`1.RedirectContainerSessionChannel.RequestAsyncResult.<>c__DisplayClass17.<GetAsyncSteps>b__a(RequestAsyncResult thisPtr, IAsyncResult r)
at Microsoft.ServiceBus.Messaging.IteratorAsyncResult`1.StepCallback(IAsyncResult result)
Exception rethrown at [2]:
at Microsoft.ServiceBus.Common.AsyncResult.End[TAsyncResult](IAsyncResult result)
at Microsoft.ServiceBus.Common.AsyncResult`1.End(IAsyncResult asyncResult)
at Microsoft.ServiceBus.Messaging.Sbmp.RedirectBindingElement.RedirectContainerChannelFactory`1.RedirectContainerSessionChannel.EndRequest(IAsyncResult result)
at Microsoft.ServiceBus.Messaging.Channels.ReconnectBindingElement.ReconnectChannelFactory`1.RequestSessionChannel.RequestAsyncResult.<GetAsyncSteps>b__4(RequestAsyncResult thisPtr, IAsyncResult r)
at Microsoft.ServiceBus.Messaging.IteratorAsyncResult`1.StepCallback(IAsyncResult result)
Exception rethrown at [3]:
at Microsoft.ServiceBus.Common.AsyncResult.End[TAsyncResult](IAsyncResult result)
at Microsoft.ServiceBus.Common.AsyncResult`1.End(IAsyncResult asyncResult)
at Microsoft.ServiceBus.Messaging.Channels.ReconnectBindingElement.ReconnectChannelFactory`1.RequestSessionChannel.EndRequest(IAsyncResult result)
at Microsoft.ServiceBus.Messaging.Sbmp.SbmpMessageReceiver.EndReceiveCommand(IAsyncResult result, IEnumerable`1& messages)
--- End of inner exception stack trace ---
at Microsoft.ServiceBus.Common.AsyncResult.End[TAsyncResult](IAsyncResult result)
at Microsoft.ServiceBus.Common.AsyncResult`1.End(IAsyncResult asyncResult)
at Microsoft.ServiceBus.Messaging.MessageReceiver.RetryReceiveAsyncResult.TryReceiveEnd(IAsyncResult r, IEnumerable`1& messages)
at Microsoft.ServiceBus.Messaging.MessageReceiver.EndTryReceive(IAsyncResult result, IEnumerable`1& messages)
at Microsoft.ServiceBus.Messaging.MessageReceiver.EndReceiveBatch(IAsyncResult result)
at System.Threading.Tasks.TaskFactory`1.FromAsyncCoreLogic(IAsyncResult iar, Func`2 endFunction, Action`1 endAction, Task`1 promise, Boolean requiresSynchronization)
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at System.Runtime.CompilerServices.TaskAwaiter`1.GetResult()
at Nimbus.Infrastructure.MessageSendersAndReceivers.NimbusQueueMessageReceiver.<FetchBatch>d__3.MoveNext() in c:\readify\Centium\NextGen\Nimbus\src\Nimbus\Infrastructure\MessageSendersAndReceivers\NimbusQueueMessageReceiver.cs:line 36
--- End of inner exception stack trace ---
at System.Threading.Tasks.Task.ThrowIfExceptional(Boolean includeTaskCanceledExceptions)
at System.Threading.Tasks.Task`1.GetResultCore(Boolean waitCompletionNotification)
at System.Threading.Tasks.Task`1.get_Result()
at Nimbus.Infrastructure.MessageSendersAndReceivers.NimbusMessageReceiver.<FetchBatchInternal>d__17.MoveNext() in c:\readify\Centium\NextGen\Nimbus\src\Nimbus\Infrastructure\MessageSendersAndReceivers\NimbusMessageReceiver.cs:line 93
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at System.Runtime.CompilerServices.TaskAwaiter`1.GetResult()
at Nimbus.Infrastructure.MessageSendersAndReceivers.NimbusMessageReceiver.<Worker>d__d.MoveNext() in c:\readify\Centium\NextGen\Nimbus\src\Nimbus\Infrastructure\MessageSendersAndReceivers\NimbusMessageReceiver.cs:line 48
---> (Inner Exception #0) Microsoft.ServiceBus.Messaging.MessagingCommunicationException: Error during communication with Service Bus. Check the connection information, then retry. ---> System.ServiceModel.CommunicationObjectFaultedException: Internal Server Error: The server did not provide a meaningful reply; this might be caused by a premature session shutdown..TrackingId:ce8706ce-c459-41a8-b6e5-5398f85055d9, Timestamp:15/04/2014 3:15:19 AM
Server stack trace:
Exception rethrown at [0]:
at Microsoft.ServiceBus.Common.AsyncResult.End[TAsyncResult](IAsyncResult result)
at Microsoft.ServiceBus.Messaging.Sbmp.DuplexRequestBindingElement.DuplexRequestSessionChannel.DuplexCorrelationAsyncResult.End(IAsyncResult result)
at Microsoft.ServiceBus.Messaging.Sbmp.DuplexRequestBindingElement.DuplexRequestSessionChannel.EndRequest(IAsyncResult result)
at Microsoft.ServiceBus.Messaging.Channels.ReconnectBindingElement.ReconnectChannelFactory`1.RequestSessionChannel.RequestAsyncResult.<GetAsyncSteps>b__4(RequestAsyncResult thisPtr, IAsyncResult r)
at Microsoft.ServiceBus.Messaging.IteratorAsyncResult`1.StepCallback(IAsyncResult result)
Exception rethrown at [1]:
at Microsoft.ServiceBus.Common.AsyncResult.End[TAsyncResult](IAsyncResult result)
at Microsoft.ServiceBus.Common.AsyncResult`1.End(IAsyncResult asyncResult)
at Microsoft.ServiceBus.Messaging.Channels.ReconnectBindingElement.ReconnectChannelFactory`1.RequestSessionChannel.EndRequest(IAsyncResult result)
at Microsoft.ServiceBus.Messaging.Sbmp.RedirectBindingElement.RedirectContainerChannelFactory`1.RedirectContainerSessionChannel.RequestAsyncResult.<>c__DisplayClass17.<GetAsyncSteps>b__a(RequestAsyncResult thisPtr, IAsyncResult r)
at Microsoft.ServiceBus.Messaging.IteratorAsyncResult`1.StepCallback(IAsyncResult result)
Exception rethrown at [2]:
at Microsoft.ServiceBus.Common.AsyncResult.End[TAsyncResult](IAsyncResult result)
at Microsoft.ServiceBus.Common.AsyncResult`1.End(IAsyncResult asyncResult)
at Microsoft.ServiceBus.Messaging.Sbmp.RedirectBindingElement.RedirectContainerChannelFactory`1.RedirectContainerSessionChannel.EndRequest(IAsyncResult result)
at Microsoft.ServiceBus.Messaging.Channels.ReconnectBindingElement.ReconnectChannelFactory`1.RequestSessionChannel.RequestAsyncResult.<GetAsyncSteps>b__4(RequestAsyncResult thisPtr, IAsyncResult r)
at Microsoft.ServiceBus.Messaging.IteratorAsyncResult`1.StepCallback(IAsyncResult result)
Exception rethrown at [3]:
at Microsoft.ServiceBus.Common.AsyncResult.End[TAsyncResult](IAsyncResult result)
at Microsoft.ServiceBus.Common.AsyncResult`1.End(IAsyncResult asyncResult)
at Microsoft.ServiceBus.Messaging.Channels.ReconnectBindingElement.ReconnectChannelFactory`1.RequestSessionChannel.EndRequest(IAsyncResult result)
at Microsoft.ServiceBus.Messaging.Sbmp.SbmpMessageReceiver.EndReceiveCommand(IAsyncResult result, IEnumerable`1& messages)
--- End of inner exception stack trace ---
at Microsoft.ServiceBus.Common.AsyncResult.End[TAsyncResult](IAsyncResult result)
at Microsoft.ServiceBus.Common.AsyncResult`1.End(IAsyncResult asyncResult)
at Microsoft.ServiceBus.Messaging.MessageReceiver.RetryReceiveAsyncResult.TryReceiveEnd(IAsyncResult r, IEnumerable`1& messages)
at Microsoft.ServiceBus.Messaging.MessageReceiver.EndTryReceive(IAsyncResult result, IEnumerable`1& messages)
at Microsoft.ServiceBus.Messaging.MessageReceiver.EndReceiveBatch(IAsyncResult result)
at System.Threading.Tasks.TaskFactory`1.FromAsyncCoreLogic(IAsyncResult iar, Func`2 endFunction, Action`1 endAction, Task`1 promise, Boolean requiresSynchronization)
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at System.Runtime.CompilerServices.TaskAwaiter`1.GetResult()
at Nimbus.Infrastructure.MessageSendersAndReceivers.NimbusQueueMessageReceiver.<FetchBatch>d__3.MoveNext() in c:\readify\Centium\NextGen\Nimbus\src\Nimbus\Infrastructure\MessageSendersAndReceivers\NimbusQueueMessageReceiver.cs:line 36<---
The EnsureXXXExists() commands on AzureQueueManager only checks if the queue/topic/subscription exists, however if the description changes after creation of the queue/topic/subscription, nimbus will not update the descriptions.
An appropriate place to check for correctness and update if required may be in the WarmUp method, after task0 and task1 have completed.
We should do a bit more sanitisation on path names we generate. Currently we have a hacky couple of calls to string.Replace(...).
Allow transactional message handlers. Maybe something like this:
https://gist.github.com/lukeschafer/6bd8551c25b3ce8c2781
This is useful for implementing functionality such as sagas. This feature is not an enabler, it purely allows for less boilerplate when implementing transactional handlers.
This will need some rework once we start allowing offline/disconnected startup but for now we just want to go bang early-on if we can't connect to the server.
When spinning up multiple instances of the same app, they compete to create new subscriptions
When we pop a message from a queue we're given a lock on that message for a short period of time. The idea is that if our handler hangs or crashes without calling .Complete or .Abandon then the message will be returned to the queue so that someone else can handle it.
There's a trade-off between 1) detecting handlers that have crashed and giving someone else a chance at handling the message and 2) allowing long-running handlers to do their job.
I'm thinking about either of:
Any thoughts? @DamianMac? @michaelnoonan?
So far, this has been exhibited running in IIS with commands already in a queue. The handler is never run but the app hangs on startup.
Nimbus's string.FormatWith
extension is public: https://github.com/NimbusAPI/Nimbus/blob/master/src/Nimbus/Extensions/StringExtensions.cs.
It's well documented that I love me a good string.FormatWith
but this one should be internal.
Working with Nimbus on a project I encountered the situation where I needed to update a database from within a handler. In this solution the front-end web project used ASP.Net MVC's Action Filter feature to create and maintain an ambient unit of work. However I don't have this option for my handlers and need to manage it myself within the "Handle" method. As I encountered this I thought this might be a useful feature for Nimbus to have.
My initial thoughts are that I'd want some method of registering a class which Nimbus could resolve within the lifetime scope that it will use to resolve and call the handler. This class will have an interface something like this.
public interface IHandlerFilter
{
public void OnPreHandle();
public void OnPostHandle();
}
This interface is obviously very basic and requires some fleshing out. I plan to do some prototype work but wanted to make sure the idea was something you guys were interested in and get some input on possible issues and direction.
I've been looking through the Nimbus source to identify the ease with which local messaging can be implemented. I don't want to move around too much infrastructural cheese, but in terms of scalability I think it's a great feature.
At the moment would I be right in saying the message flow (one-way) is roughly as follows (sorry about the crummy text-diagrams):
[Bus] -> Execute operation to -> [Sender (eg. ICommandSender)] -> Get queue (IMessageSender) from -> [MessagingFactory] -> Create queue if necessary in -> [QueueManager] -> Send message to queue <- Gets messages <- [Message Pump] -> Sends to -> [Dispatcher] -> Creates instance of -> [Handler] -> Handles
So in short:
Outgoing:
[Bus] -> [Sender] -> [MessagingFactory] -> Queue
Incoming:
Queue -> [MessagePump] -> [Dispatcher] -> [Handler]
My early thoughts are to wrap an F# Mailbox Processor in the IMessageSender and IMessageReceiver interfaces and for each Azure Queue that is created, create an equivalent local queue.
When the MessagingFactory passes the queue to the sender it will first try to find a local queue. If it does, and the Bus has opted in to local messaging, return it over the Azure queue. If not, return the Azure queue. It would look like this:
Outgoing:
[Bus] -> [Sender] -> [MessagingFactory] -> Local or External? -> Queue
Incoming (Local):
Queue -> [Dispatcher] -> [Handler]
Incoming (External):
Azure Queue -> [Message Pump] -> Local Queue -> [Dispatcher] -> [Handler]
In the message pump, route all Azure Queue traffic through the equivalent local queue which will then call the dispatcher, rather than calling the dispatcher through the pump. This way, if you have multiple services running on the same machine you can optimise bandwidth and performance by messaging via the F# queues instead of Azure, but also receive and handle traffic from the Azure queues in exactly the same way.
I think the important things to note would be: local messaging is less resilient, so it would be opt in to return local queues. Whether or not the Bus opts in, still build the local queues and route MessagePump traffic through them anyway because this makes no difference in that case.
Does anyone have anything to add or potential caveats they can see? I'm playing around with the concept locally in a fork. If I implement this without breaking existing funcitonality would it be pulled into the main branch or is it outside the scope of Nimbus?
Delayed abandons would be useful for retry backoffs. E.g. instead of
await message.AbandonAsync(exception.ExceptionDetailsAsProperties(_clock.UtcNow));
you can
If (_settings.DelayedAbandon.HasValue) Thread.Sleep(_settings.DelayedAbandon.GetValueOrDefault());
await message.AbandonAsync(exception.ExceptionDetailsAsProperties(_clock.UtcNow));
It's safe (durable) to wait in-proc like this as a) it's already on its own thread, and b) if the process dies the peeklock will simply expire and the message will become available again
Optional:
Exponential backoffs - when abandoning a message, use DeliveryCount, or add the last-waited/abandon count to the message properties and read this when backing off.
Happy to implement everything above if you agree.
I think it makes sense for TC to do a git clean before building. I pulled the latest source code and a clean build failed due to the build order: Nimbus.IntegrationTests built before the Autofac and Windsor IntegrationTest libraries.
I've committed a fix for the project build ordering in 2c1b9f1, but TC should have picked this up earlier in my opinion.
public class MyHandler : IHandleCommand<MyMessage>
{
public async void Handle(MyMessage busCommand)
{
throw new InvalidOperationException("HA HA HA, you can't catch me!");
}
}
This message will be viewed as handled, and therefore not retried/put in error queue.
Perhaps Nimbus should enforce (throw upfront) when I have async implementation of the Handle
method?
It's easy to miss (we did) and can cause a lot of pain.
Discovered by @DamianMac. We have race conditions when creating multiple bus instances in separate processes (e.g. when hitting F5 on a solution with multiple startup projects) because we ensure the existence of queues and topics by checking whether they exist and then creating them if not.
Considering just implementing this as a data-driven test for multiple instances of IBus. Now that we have message brokers for Activator.CreateInstance, Autofac and Windsor we need to be confident that behaviour across all of them is consistent.
A declarative, efficient, and flexible JavaScript library for building user interfaces.
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ๐๐๐
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google โค๏ธ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.