Initial ConsumeSingle implementation.
Merge pull request #1 from mikehadlow/consume-single

Consume single from hadlow
A bit of fun with reflection

I found a usage of costly reflection functions: GetCustomAttributes and
Activator.CreateInstance(http://msdn.microsoft.com/en-us/magazine/cc163759.aspx#S1).

Here is an attempt to cache reflection calls. Also, perfomance tests are
included.
Fix bug and cache with static initialization
1. Split IPublishExchangeDeclareStrategy in two..
  * PublishExchangeDeclareStrategy was referring directly to the container. Moved to constructor
  * Split interface in two: An advanced and a convention based one.
Extract conventions from Rpc/restructure.
	a) I don't agree with the request/response implementation
	  - There can only be one rpc-service for each (Type x Type).
	  - The responseQueue handling is not necessary.
	  It implies that a service can only have 1 consumer for each (Type x Type), since their keys would conflict.
	  (So in that sense it is really unnecessary that it is a Concurrent one)
	  - For the same reason the request-queue is exclusive. This can cause downtime if an old connection is not timed out when restarting a service.

	EasyNetQ should be able to support several servers IMO -- all the hard work is already done in rabbit.
	--
	On rabbit this rpc pattern is usually done with a random queue per response. Set expires (a bit larger than the timeout), so that we will not cloud the rabbit server.

	Here is the high-level scheme I have used before

	0. The server declares a request-queue which subscribes to some routingkey.
	1. The client creates a response queue with random name. Sets the expires to something more than the response timeout.
	2. The client decides the response routingkey (could just be the response-queue-name), and puts it in the request message header.
	3. The client publishes the request message
	4. The server receives the request, calls the handler, and gets the response.
	5. The server extracts the response routingkey from the request-header, and publishes the response message (maybe also propagate a correlation id)

	Then we don't need the response queue management and the response-queue management is kept local in the handler method.

	b) EasyNetQ is using a direct exchange (the default) for the rpc. It makes sense for the response queues, but I think it is a real pity for the requests.

	If the request publish happens in a topic exchange, we can have different contexts (topics) of rpc for the same request/reply types.
	Other services might be interested in the messages as well. For logging, monitoring, etc.

	c) I don't see why the client and the server implementation is mixed to such an extent. I have split it in two, since it started to confuse me.

	d) Added handler-id (subscription-id) to respond -- just like it is possible for IBus.Subscribe(). Added a generator, RpcRequestQueueNameConvention, in IConvention
	   Implemented like:
	     RpcRequestQueueNameConvention = (type, id) => "Rpc:"+QueueNamingConvention(type,id);

	e) An rpc request queue should expire+autodelete. It doesn't make sense to handle an old message. The client times out anyway.

	f) Due to e we want to cancel the response consume when we have consumed 1 message. I would like you to take a look at this one:

		class AdvancedClientRpc {

			//...

			private void SubscribeToResponse(string responseQueueName)
			{
				//...
				//the response is published to the default exchange with the queue name as routingkey. So no need to bind to exchange
				var disposeWhenSet = new DisposeWhenSet();
				var consumingDisposer = _advancedBus.Consume(queue, (bytes, msgProp, messageReceivedInfo) => Task.Factory.StartNew(() =>
				{
					ResponseAction responseAction;
					if (responseActions.TryRemove(msgProp.CorrelationId, out responseAction))
					{
						responseAction.OnSuccess(new SerializedMessage(msgProp, bytes));
					}
					disposeWhenSet.DisposeObject();
				}));
				disposeWhenSet.Disposable = consumingDisposer;
				//...
			}
			//...
		}

		I needed to add a wrapper for disposing the consumption. But it stinks like something that can cause a dead-lock. (The dispose call waiting for the task to finish.)
		Looking at InternalConsumer.Dispose() though..
			public void Dispose()
			{
				if (disposed) return;
				disposed = true;

				var model = Model;
				if (model != null)
				{
					// Queued because we may be on the RabbitMQ.Client dispatch thread.
					consumerDispatcher.QueueAction(() =>
						{
							Model.Dispose();
							eventBus.Publish(new ConsumerModelDisposedEvent(ConsumerTag));
						});
				}
			}
		... it seems to be safe.
		It is probably much better to add a ConsumeSingle to the advanced api.

IRpc and Rpc is unchanged. So the solution is backwards compatible. I have made another implementation BetterRpc (Feel free to rename), which uses the new scheme. The two implementation are switchable in ComponentRegistration
Added advanced rpc to IAdvancedBus + Moved rpc stuff to own folder

Conflicts:
	Source/EasyNetQ/IAdvancedBus.cs
	Source/EasyNetQ/RabbitAdvancedBus.cs
Moved rpc stuff to own folder

Conflicts:
	Source/EasyNetQ/Producer/Rpc.cs
Moved rpc classes to their own files
Fix stackoverflow in initialization.

I turned out to have a circular dependency which crashed the service locator (stackoverflow). It seemed like such a simple commit :-)
I had to introduce the IAdvancedRpcFactory since the implementation needs the advanced bus.

    interface IAdvancedRpcFactory
    {
        IAdvancedClientRpc CreateClientRpc(IAdvancedBus advancedBus);
        IAdvancedServerRpc CreateServerRpc(IAdvancedBus advancedBus);
    }

But I guess it makes sense that they are defined at the same time. Then they are more obviously compatible

Conflicts:
	Source/EasyNetQ/RabbitAdvancedBus.cs
Fixed bug: Exception propagation from server to client.
Reverted to .Register<IRpc, Rpc.Rpc>()

One can still plug the new implementation. But since there are semantical changes, the tests don't pass.
1. Split IPublishExchangeDeclareStrategy in two..
  * PublishExchangeDeclareStrategy was referring directly to the container. Moved to constructor
  * Split interface in two: An advanced and a convention based one.
Extract conventions from Rpc/restructure.
	a) I don't agree with the request/response implementation
	  - There can only be one rpc-service for each (Type x Type).
	  - The responseQueue handling is not necessary.
	  It implies that a service can only have 1 consumer for each (Type x Type), since their keys would conflict.
	  (So in that sense it is really unnecessary that it is a Concurrent one)
	  - For the same reason the request-queue is exclusive. This can cause downtime if an old connection is not timed out when restarting a service.

	EasyNetQ should be able to support several servers IMO -- all the hard work is already done in rabbit.
	--
	On rabbit this rpc pattern is usually done with a random queue per response. Set expires (a bit larger than the timeout), so that we will not cloud the rabbit server.

	Here is the high-level scheme I have used before

	0. The server declares a request-queue which subscribes to some routingkey.
	1. The client creates a response queue with random name. Sets the expires to something more than the response timeout.
	2. The client decides the response routingkey (could just be the response-queue-name), and puts it in the request message header.
	3. The client publishes the request message
	4. The server receives the request, calls the handler, and gets the response.
	5. The server extracts the response routingkey from the request-header, and publishes the response message (maybe also propagate a correlation id)

	Then we don't need the response queue management and the response-queue management is kept local in the handler method.

	b) EasyNetQ is using a direct exchange (the default) for the rpc. It makes sense for the response queues, but I think it is a real pity for the requests.

	If the request publish happens in a topic exchange, we can have different contexts (topics) of rpc for the same request/reply types.
	Other services might be interested in the messages as well. For logging, monitoring, etc.

	c) I don't see why the client and the server implementation is mixed to such an extent. I have split it in two, since it started to confuse me.

	d) Added handler-id (subscription-id) to respond -- just like it is possible for IBus.Subscribe(). Added a generator, RpcRequestQueueNameConvention, in IConvention
	   Implemented like:
	     RpcRequestQueueNameConvention = (type, id) => "Rpc:"+QueueNamingConvention(type,id);

	e) An rpc request queue should expire+autodelete. It doesn't make sense to handle an old message. The client times out anyway.

	f) Due to e we want to cancel the response consume when we have consumed 1 message. I would like you to take a look at this one:

		class AdvancedClientRpc {

			//...

			private void SubscribeToResponse(string responseQueueName)
			{
				//...
				//the response is published to the default exchange with the queue name as routingkey. So no need to bind to exchange
				var disposeWhenSet = new DisposeWhenSet();
				var consumingDisposer = _advancedBus.Consume(queue, (bytes, msgProp, messageReceivedInfo) => Task.Factory.StartNew(() =>
				{
					ResponseAction responseAction;
					if (responseActions.TryRemove(msgProp.CorrelationId, out responseAction))
					{
						responseAction.OnSuccess(new SerializedMessage(msgProp, bytes));
					}
					disposeWhenSet.DisposeObject();
				}));
				disposeWhenSet.Disposable = consumingDisposer;
				//...
			}
			//...
		}

		I needed to add a wrapper for disposing the consumption. But it stinks like something that can cause a dead-lock. (The dispose call waiting for the task to finish.)
		Looking at InternalConsumer.Dispose() though..
			public void Dispose()
			{
				if (disposed) return;
				disposed = true;

				var model = Model;
				if (model != null)
				{
					// Queued because we may be on the RabbitMQ.Client dispatch thread.
					consumerDispatcher.QueueAction(() =>
						{
							Model.Dispose();
							eventBus.Publish(new ConsumerModelDisposedEvent(ConsumerTag));
						});
				}
			}
		... it seems to be safe.
		It is probably much better to add a ConsumeSingle to the advanced api.

IRpc and Rpc is unchanged. So the solution is backwards compatible. I have made another implementation BetterRpc (Feel free to rename), which uses the new scheme. The two implementation are switchable in ComponentRegistration

Conflicts:
	Source/EasyNetQ/ComponentRegistration.cs
Added advanced rpc to IAdvancedBus + Moved rpc stuff to own folder

Conflicts:
	Source/EasyNetQ/IAdvancedBus.cs
	Source/EasyNetQ/RabbitAdvancedBus.cs
	Source/EasyNetQ/Rpc/BetterRpc.cs
Moved rpc stuff to own folder

Conflicts:
	Source/EasyNetQ/Producer/Rpc.cs
	Source/EasyNetQ/Rpc/IRpc.cs
Moved rpc classes to their own files

Conflicts:
	Source/EasyNetQ/Rpc/IRpc.cs
Clean up from Merge consume single
Added TODO comment. DeclareExchange optimization is not semantically correct.
Fresh rpc strategy seems to work