Mit der Mai CTP von Azure AppFabric [1] stellt Microsoft erstmals eine Implementierung von Topics zur Verfügung – wenn auch vorerst lediglich über die Azure-Cloud. Über ein REST-Interface oder einen .NET-Wrapper können damit Nachrichten an Abonnenten versendet werden. Darüber hinaus besteht auch die Möglichkeit, Topics direkt über WCF zu verwenden. Bei solchen Szenarien nehmen WCF-Services die Rolle der Abonnenten ein und WCF-Clients die Rolle des Verlegers.
Das nachfolgende Beispiel demonstriert dies. Dazu kommt ein Topic FlugBuchungenTopic mit zwei Abonnenten zum Einsatz: Accounting und BonusMilesManager. Letzterer weist auch einen Filter auf. Dieser schließt Flugbuchungen, die nicht von Vielfliegern stammen, aus. Die zum Filtern benötigten Daten müssen allerdings zusätzlich in Form von Nachrichteneigenschaften übertragen werden. Deswegen kommt beim Senden der OperationContext zum Einsatz, um diese Metadaten als Teil einer Instanz von BrokeredMessageProperty zu hinterlegen. Die Service-Operation ruft am Begin diese Instanz, welche in der Auflistung IncomingMessageProperties zu finden sind, ab.
Topic und Abos anlegen
private void CreateIfNotExist(string suffix)
{
SharedSecretCredential myManagementCredentials =
TransportClientCredentialBase.CreateSharedSecretCredential(IssuerName, IssuerKey);
ServiceBusNamespaceClient namespaceClient = new ServiceBusNamespaceClient(
ServiceBusEnvironment.CreateServiceUri("sb", ServiceNamespace, string.Empty),
myManagementCredentials);
string topicName = "FlugBuchungenTopic" + suffix;
Topic myTopic;
myTopic = namespaceClient.GetTopics().FirstOrDefault(t => t.Path == topicName.ToLower());
if (myTopic == null)
{
myTopic = namespaceClient.CreateTopic(topicName.ToLower());
}
var subs = myTopic.GetSubscriptions();
if (subs.Count(s => s.Name == "Accounting".ToLower()) == 0)
{
SubscriptionDescription desc;
desc = new SubscriptionDescription();
desc.EnableDeadLetteringOnMessageExpiration = true;
RuleDescription ruleDesc;
ruleDesc = new RuleDescription();
ruleDesc.FilterExpression = new SqlFilterExpression("IstVielFlieger = 'True'");
ruleDesc.FilterAction = new SqlFilterAction("set PremiumKunde = 'True'");
var acc = myTopic.AddSubscription("Accounting".ToLower(), desc /*, ruleDesc*/);
}
if (subs.Count(s => s.Name == "BonusMilesManager".ToLower()) == 0)
{
SubscriptionDescription desc;
desc = new SubscriptionDescription();
desc.EnableDeadLetteringOnMessageExpiration = true;
var filter = new SqlFilterExpression("IstVielFlieger = 'True'");
myTopic.AddSubscription("BonusMilesManager".ToLower(), desc, filter);
}
}
Service-Vertrag
[ServiceContract()]
public interface IFlugBuchungService
{
[OperationContract(IsOneWay = true)]
void Buche(FlugBuchung b);
}
Client
IFlugBuchungService proxy;
ChannelFactory<IFlugBuchungService> cf = new ChannelFactory<IFlugBuchungService>("FlugBuchungWcfTopic");
proxy = cf.CreateChannel();
for (int i = 0; i < 10; i++)
{
using (new OperationContextScope((IContextChannel) proxy))
{
FlugBuchung fb = new FlugBuchung();
fb.Datum = DateTime.Now;
fb.FlugNummer = "LH4711";
fb.Von = "Graz";
fb.Nach = "Mallorca";
fb.Vorname = "Maximilian";
fb.Nachname = "Muster";
fb.VielfliegerNummer = "LH-0000-0000-000" + i;
var bmp = new BrokeredMessageProperty();
bmp.Properties["IstVielFlieger"] = (i % 2 == 0) ? "True" : "False";
bmp.TimeToLive = TimeSpan.FromMinutes(5);
OperationContext.Current.OutgoingMessageProperties.Add(BrokeredMessageProperty.Name, bmp);
proxy.Buche(fb);
}
}
Service-Implementierung
public class FlugBuchungService: IFlugBuchungService
{
public void Buche(FlugBuchung b)
{
string istVielFlieger = "";
var obj = OperationContext.Current.IncomingMessageProperties[BrokeredMessageProperty.Name];
var property = obj as BrokeredMessageProperty;
if (property != null) istVielFlieger = property.Properties["IstVielFlieger"].ToString();
Console.WriteLine(b);
Console.WriteLine("IstVielflieger: " + istVielFlieger);
Console.WriteLine();
}
}
Konfiguration (Service und Client)
<configuration>
<system.serviceModel>
<extensions>
<bindingElementExtensions>
<add name="serviceBusMessagingTransport" type="Microsoft.ServiceBus.Messaging.Configuration.ServiceBusMessagingTransportExtensionElement,
Microsoft.ServiceBus.Messaging, Version=2.0.0.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35"/>
</bindingElementExtensions>
<bindingExtensions>
<add name="serviceBusMessagingBinding" type="Microsoft.ServiceBus.Messaging.Configuration.ServiceBusMessagingBindingCollectionElement,
Microsoft.ServiceBus.Messaging, Version=2.0.0.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35"/>
</bindingExtensions>
<!--
<behaviorExtensions>
<add name="transportClientEndpointBehavior" type="Microsoft.ServiceBus.Configuration.TransportClientEndpointBehaviorElement, Microsoft.ServiceBus, Version=2.0.0.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35"/>
</behaviorExtensions>
-->
</extensions>
<behaviors>
<endpointBehaviors>
<behavior name="securityBehavior">
<transportClientEndpointBehavior>
<clientCredentials>
<sharedSecret
issuerName="owner"
issuerSecret="LVwY06Kx0Q5rBfDEXByBuXuaHIlXk92qupmdpxqF4V4="/>
</clientCredentials>
</transportClientEndpointBehavior>
</behavior>
</endpointBehaviors>
<serviceBehaviors>
<behavior>
<serviceDebug includeExceptionDetailInFaults="true" />
<serviceMetadata httpGetEnabled="true"/>
</behavior>
</serviceBehaviors>
</behaviors>
<bindings>
<serviceBusMessagingBinding>
<binding name="serviceBusBinding" closeTimeout="00:03:00" openTimeout="00:03:00"
receiveTimeout="00:03:00" sendTimeout="00:03:00" >
<messagingFactorySettings>
<sbmpTransport batchFlushInterval="00:00:01"/>
</messagingFactorySettings>
</binding>
</serviceBusMessagingBinding>
</bindings>
<client>
<endpoint name="FlugBuchungWcfQueue"
address="sb://steyer-test.servicebus.appfabriclabs.com/flugbuchungentopicwcf/"
binding="serviceBusMessagingBinding"
bindingConfiguration="serviceBusBinding"
contract="WCF_Queue_Service.IFlugBuchungService"
behaviorConfiguration="securityBehavior"/>
</client>
<services>
<service name="WCF_Queue_Service.FlugBuchungService">
<endpoint name="FlugBuchungWcfQueueEP"
address="sb://steyer-test.servicebus.appfabriclabs.com/flugbuchungentopicwcf"
listenUri="sb://steyer-test.servicebus.appfabriclabs.com/flugbuchungentopicwcf/subscriptions/accounting"
binding="serviceBusMessagingBinding"
bindingConfiguration="serviceBusBinding"
contract="WCF_Queue_Service.IFlugBuchungService"
behaviorConfiguration="securityBehavior" />
<host>
<baseAddresses>
<add baseAddress="http://localhost:8080/FlugBuchungService"/>
</baseAddresses>
</host>
</service>
</services>
</system.serviceModel>
</configuration>
Feed
Twitter
XING
