1
0
mirror of synced 2024-11-21 17:03:50 +00:00

Some more interface foundations

This commit is contained in:
Mark van Renswoude 2016-12-05 23:41:17 +01:00
parent 039fd3d6f2
commit 3e27d56809
10 changed files with 218 additions and 64 deletions

View File

@ -16,7 +16,7 @@ namespace Tapeti.Connection
private readonly IMessageSerializer messageSerializer;
private readonly IRoutingKeyStrategy routingKeyStrategy;
private readonly Lazy<SingleThreadTaskQueue> taskQueue = new Lazy<SingleThreadTaskQueue>();
private IConnection connection;
private RabbitMQ.Client.IConnection connection;
private IModel channel;

9
IConnection.cs Normal file
View File

@ -0,0 +1,9 @@
using System;
namespace Tapeti
{
public interface IConnection : IDisposable
{
ISubscriber Subscribe();
}
}

20
ITopology.cs Normal file
View File

@ -0,0 +1,20 @@
using System.Collections.Generic;
namespace Tapeti
{
public interface ITopology
{
IEnumerable<IQueue> Queues();
}
public interface IQueue
{
IEnumerable<IBinding> Bindings();
}
public interface IBinding
{
}
}

View File

@ -59,10 +59,13 @@
<Compile Include="Connection\TapetiWorker.cs" />
<Compile Include="Default\ConsoleLogger.cs" />
<Compile Include="Default\DevNullLogger.cs" />
<Compile Include="IConnection.cs" />
<Compile Include="ILogger.cs" />
<Compile Include="ITopology.cs" />
<Compile Include="MessageController.cs" />
<Compile Include="TapetiConnectionExtensions.cs" />
<Compile Include="TapetiConnectionBuilder.cs" />
<Compile Include="TapetiConnectionParams.cs" />
<Compile Include="TapetiTopologyBuilder.cs" />
<Compile Include="TapetiTypes.cs" />
<Compile Include="Tasks\SingleThreadTaskQueue.cs" />
<Compile Include="Default\DefaultControllerFactory.cs" />

View File

@ -60,37 +60,6 @@ namespace Tapeti
}
public TapetiConnection RegisterController(Type type)
{
var queueAttribute = type.GetCustomAttribute<MessageController>();
if (queueAttribute == null)
throw new ArgumentException("Queue attribute required on class", nameof(type));
if (queueAttribute.Dynamic)
{
if (!string.IsNullOrEmpty(queueAttribute.Name))
throw new ArgumentException("Dynamic queue attributes must not have a Name");
registrations.Value.Add(new ControllerDynamicQueueRegistration(
DependencyResolver.Resolve<IControllerFactory>,
DependencyResolver.Resolve<IRoutingKeyStrategy>,
type, SubscribeExchange));
}
else
{
if (string.IsNullOrEmpty(queueAttribute.Name))
throw new ArgumentException("Non-dynamic queue attribute must have a Name");
registrations.Value.Add(new ControllerQueueRegistration(
DependencyResolver.Resolve<IControllerFactory>,
type, SubscribeExchange, queueAttribute.Name));
}
(DependencyResolver as IDependencyInjector)?.RegisterController(type);
return this;
}
public async Task<ISubscriber> Subscribe()
{
if (!registrations.IsValueCreated || registrations.Value.Count == 0)

View File

@ -0,0 +1,30 @@
using System;
namespace Tapeti
{
public class TapetiConnectionBuilder
{
public IConnection Build()
{
throw new NotImplementedException();
}
public TapetiConnectionBuilder SetExchange(string exchange)
{
return this;
}
public TapetiConnectionBuilder SetDependencyResolver(IDependencyResolver dependencyResolver)
{
return this;
}
public TapetiConnectionBuilder SetTopology(ITopology topology)
{
return this;
}
}
}

View File

@ -1,23 +0,0 @@
using System.Linq;
using System.Reflection;
using Tapeti.Annotations;
namespace Tapeti
{
public static class TapetiConnectionExtensions
{
public static TapetiConnection RegisterAllControllers(this TapetiConnection connection, Assembly assembly)
{
foreach (var type in assembly.GetTypes().Where(t => t.IsDefined(typeof(DynamicQueueAttribute))))
connection.RegisterController(type);
return connection;
}
public static TapetiConnection RegisterAllControllers(this TapetiConnection connection)
{
return RegisterAllControllers(connection, Assembly.GetCallingAssembly());
}
}
}

146
TapetiTopologyBuilder.cs Normal file
View File

@ -0,0 +1,146 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using Tapeti.Annotations;
namespace Tapeti
{
public class TopologyConfigurationException : Exception
{
public TopologyConfigurationException(string message) : base(message) { }
}
public class TapetiTopologyBuilder
{
private readonly List<HandlerRegistration> registrations = new List<HandlerRegistration>();
public ITopology Build()
{
throw new NotImplementedException();
}
public TapetiTopologyBuilder RegisterController(Type controller)
{
var controllerRegistration = GetAttributesRegistration(controller);
foreach (var method in controller.GetMembers(BindingFlags.Public | BindingFlags.Instance)
.Where(m => m.MemberType == MemberTypes.Method && m.DeclaringType != typeof(object))
.Select(m => (MethodInfo)m))
{
}
/*
if (queueAttribute.Dynamic)
{
if (!string.IsNullOrEmpty(queueAttribute.Name))
throw new ArgumentException("Dynamic queue attributes must not have a Name");
registrations.Value.Add(new ControllerDynamicQueueRegistration(
DependencyResolver.Resolve<IControllerFactory>,
DependencyResolver.Resolve<IRoutingKeyStrategy>,
type, SubscribeExchange));
}
else
{
if (string.IsNullOrEmpty(queueAttribute.Name))
throw new ArgumentException("Non-dynamic queue attribute must have a Name");
registrations.Value.Add(new ControllerQueueRegistration(
DependencyResolver.Resolve<IControllerFactory>,
type, SubscribeExchange, queueAttribute.Name));
}
(DependencyResolver as IDependencyInjector)?.RegisterController(type);
*/
return this;
}
public TapetiTopologyBuilder RegisterAllControllers(Assembly assembly)
{
foreach (var type in assembly.GetTypes().Where(t => t.IsDefined(typeof(MessageControllerAttribute))))
RegisterController(type);
return this;
}
public TapetiTopologyBuilder RegisterAllControllers()
{
return RegisterAllControllers(Assembly.GetCallingAssembly());
}
protected HandlerRegistration GetAttributesRegistration(MemberInfo member)
{
var registration = new HandlerRegistration();
var dynamicQueueAttribute = member.GetCustomAttribute<DynamicQueueAttribute>();
var staticQueueAttribute = member.GetCustomAttribute<StaticQueueAttribute>();
if (dynamicQueueAttribute != null && staticQueueAttribute != null)
throw new TopologyConfigurationException($"Cannot combine static and dynamic queue attributes on {member.Name}");
if (dynamicQueueAttribute != null)
registration.Dynamic = true;
else if (staticQueueAttribute != null)
{
registration.Dynamic = false;
registration.QueueName = staticQueueAttribute.Name;
}
return registration;
}
protected class HandlerRegistration
{
public bool? Dynamic { get; set; }
public string QueueName { get; set; }
}
protected class Topology : ITopology
{
private readonly List<Queue> queues = new List<Queue>();
public void Add(Queue queue)
{
queues.Add(queue);
}
public IEnumerable<IQueue> Queues()
{
return queues;
}
}
protected class Queue : IQueue
{
private readonly List<Binding> bindings = new List<Binding>();
public void Add(Binding binding)
{
bindings.Add(binding);
}
public IEnumerable<IBinding> Bindings()
{
return bindings;
}
}
protected class Binding : IBinding
{
}
}
}

View File

@ -11,13 +11,14 @@ namespace Test
{
var container = new Container();
using (var connection = new TapetiConnection
{
PublishExchange = "test",
SubscribeExchange = "test"
}
.WithDependencyResolver(new SimpleInjectorDependencyResolver(container))
.RegisterAllControllers(typeof(Program).Assembly))
using (var connection = new TapetiConnectionBuilder()
.SetExchange("test")
.SetDependencyResolver(new SimpleInjectorDependencyResolver(container))
.SetTopology(
new TapetiTopologyBuilder()
.RegisterAllControllers()
.Build())
.Build())
{
container.Register<MarcoEmitter>();

View File

@ -1,5 +1,4 @@
using System;
using System.Threading.Tasks;
using Tapeti;
using Tapeti.Annotations;