forked from hibernating-rhinos/rhino-esb
-
Notifications
You must be signed in to change notification settings - Fork 10
/
AbstractRhinoServiceBusFacility.cs
155 lines (125 loc) · 5.48 KB
/
AbstractRhinoServiceBusFacility.cs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
using System;
using System.Collections.Generic;
using System.Configuration;
using System.Linq;
using Castle.Core;
using Castle.MicroKernel;
using Castle.MicroKernel.Facilities;
using Castle.MicroKernel.Registration;
using Castle.MicroKernel.Resolvers.SpecializedResolvers;
using Rhino.ServiceBus.Config;
using Rhino.ServiceBus.Exceptions;
using Rhino.ServiceBus.Internal;
using Rhino.ServiceBus.MessageModules;
using Rhino.ServiceBus.Msmq;
using Rhino.ServiceBus.Sagas;
using Rhino.ServiceBus.Serializers;
using System.Transactions;
namespace Rhino.ServiceBus.Impl
{
public abstract class AbstractRhinoServiceBusFacility : AbstractFacility
{
protected readonly List<Type> messageModules = new List<Type>();
private Type serializerImpl = typeof(XmlMessageSerializer);
protected IsolationLevel queueIsolationLevel = IsolationLevel.Serializable;
public bool consumeInTxn = true;
protected AbstractRhinoServiceBusFacility()
{
ThreadCount = 1;
NumberOfRetries = 5;
Transactional = TransactionalOptions.FigureItOut;
}
public Uri Endpoint { get; set; }
public int NumberOfRetries { get; set; }
public int ThreadCount { get; set; }
public bool UseFlatQueue { get; set; }
public bool DisableAutoQueueCreation { get; set; }
public TransactionalOptions Transactional { get; set; }
public IsolationLevel IsolationLevel
{
get { return queueIsolationLevel; }
}
public bool ConsumeInTransaction
{
get { return consumeInTxn; }
}
public AbstractRhinoServiceBusFacility AddMessageModule<TModule>()
where TModule : IMessageModule
{
messageModules.Add(typeof(TModule));
return this;
}
public AbstractRhinoServiceBusFacility InsertMessageModuleAtFirst<TModule>()
where TModule : IMessageModule
{
messageModules.Insert(0, typeof (TModule));
return this;
}
protected override void Init()
{
if(FacilityConfig==null)
throw new ConfigurationErrorsException(
"could not find facility configuration section with the same name of the facility");
Kernel.ComponentModelCreated += Kernel_OnComponentModelCreated;
Kernel.Resolver.AddSubResolver(new ArrayResolver(Kernel));
ReadConfiguration();
Kernel.Register(
AllTypes.FromAssembly(typeof(IBusConfigurationAware).Assembly)
.BasedOn<IBusConfigurationAware>()
);
foreach (var configurationAware in Kernel.ResolveAll<IBusConfigurationAware>())
{
configurationAware.Configure(this, FacilityConfig);
}
foreach (var type in messageModules)
{
if (Kernel.HasComponent(type) == false)
Kernel.Register(Component.For(type).Named(type.FullName));
}
RegisterComponents();
Kernel.Register(
Component.For<IReflection>()
.LifeStyle.Is(LifestyleType.Singleton)
.ImplementedBy<DefaultReflection>(),
Component.For<IMessageSerializer>()
.LifeStyle.Is(LifestyleType.Singleton)
.ImplementedBy(serializerImpl),
Component.For<IEndpointRouter>()
.ImplementedBy<EndpointRouter>()
);
}
protected abstract void RegisterComponents();
protected abstract void ReadConfiguration();
private static void Kernel_OnComponentModelCreated(ComponentModel model)
{
if (typeof(IMessageConsumer).IsAssignableFrom(model.Implementation) == false)
return;
var interfaces = model.Implementation.GetInterfaces()
.Where(x => x.IsGenericType && x.IsGenericTypeDefinition == false)
.Select(x => x.GetGenericTypeDefinition())
.ToList();
if (interfaces.Contains(typeof(InitiatedBy<>)) &&
interfaces.Contains(typeof(ISaga<>)) == false)
{
throw new InvalidUsageException("Message consumer: " + model.Implementation + " implements InitiatedBy<TMsg> but doesn't implment ISaga<TState>. " + Environment.NewLine +
"Did you forget to inherit from ISaga<TState> ?");
}
if (interfaces.Contains(typeof(InitiatedBy<>)) == false &&
interfaces.Contains(typeof(Orchestrates<>)))
{
throw new InvalidUsageException("Message consumer: " + model.Implementation + " implements Orchestrates<TMsg> but doesn't implment InitiatedBy<TState>. " + Environment.NewLine +
"Did you forget to inherit from InitiatedBy<TState> ?");
}
model.LifestyleType = LifestyleType.Transient;
}
public void UseMessageSerializer<TMessageSerializer>()
{
serializerImpl = typeof(TMessageSerializer);
}
public IFacility DisableQueueAutoCreation()
{
DisableAutoQueueCreation = true;
return this;
}
}
}