A different rpc implementation + add rpc to IAdvancedBus

Hi Mike,

Here is what I managed to do yesterday and today.

What we talked about in https://github.com/mikehadlow/EasyNetQ/pull/227

You might want to look at it before you merge anything else to master :-)

The commit timestamps are obviously not correct. I ran into some trouble, so I applied all the patches again. I hope I managed it without anything missing.

-- Steff

--------------------------
# Overall Work Log

## General theme
Split in two: An advanced and a convention based implementation.
The convention one uses the advanced interface. The advanced handles the general logic, the convention-impl adds the conventions (and nothing else -- or close to)

Some splits I have kept in the same file, so it is easier for you to approve. Up until the move commit.


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. 
 
  
2. 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 Advanced Rpc can be reached from the IAdvancedBus.
BetterRpc is implemented by means of IAdvancedBus.RequestAsync and IAdvancedBus.Respond

---

Moved all the rpc related stuff to own folder

---

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

---
This is a race condition. Once the task completes the HandlerRunner will attempt to send a Basic ACK to RabbitMQ over the consumer owned IModel (channel). At the same time you are calling dispose on the consumer which disposes of the model. They are invoked on different threads, so it's a race to see which one completes first.