add global request and response filter

Author: onlyann

README.md

@@ -47,6 +47,8 @@ The [AzureBusServer](src\ServiceStack.AzureServiceBus\AzureBusServer.cs) has the
 - `int` **RetryCount** - How many times a message should be retried before sending to the DLQ.
 - `string` **connectionString** - The connection string to the Azure Service Bus namespace
 - `IMessageFactory` **MessageFactory** - the MQ Message Factory used by this MQ Server
+- `Func<IMessage, IMessage>` **RequestFilter** - Execute global transformation or custom logic before a request is processed. Must be thread-safe.
+- `Func<object, object>` **ResponseFilter** - Execute global transformation or custom logic on the response. Must be thread-safe.
 - `Action<QueueDescription>` **CreateQueueFilter** - A filter to customize the options Azure Queues are created/updated with.
 - `Action<string, BrokeredMessage>` **GetMessageFilter** - Called every time a message is received.
 - `Action<string, BrokeredMessage, IMessage>` **PublishMessageFilter** - Called every time a message gets published.
@@ -66,9 +68,7 @@ mqServer.RegisterHandler<Hello>(m => { .. }, noOfThreads:4);
 
 > Behind the scenes, it delegates the work to Azure Service Bus [event-driven message pump](https://docs.microsoft.com/en-us/dotnet/api/microsoft.servicebus.messaging.messagereceiver.onmessage?view=azureservicebus-4.1.1#Microsoft_ServiceBus_Messaging_MessageReceiver_OnMessage_System_Action_Microsoft_ServiceBus_Messaging_BrokeredMessage__Microsoft_ServiceBus_Messaging_OnMessageOptions_).
 
-### Filters
-
-#### Create Queue Filter
+### Create Queue Filter
 
 To modify the options a queue gets created with, provide a `CreateQueueFilter` filter and modify the `QueueDescription`.
 
@@ -83,7 +83,7 @@ For instance, to change the default TTL and have messages expire automatically a
 
 If the queue already exists, any change to the queue options will result in an update.
 
-#### Message Filters
+### Message Filters
 
 There are optional `PublishMessageFilter` and `GetMessageFilter` callbacks which can be used to intercept outgoing and incoming messages. The Type name of the message body that was published is available in the `Label` property, e.g:
 
@@ -111,5 +111,4 @@ Note that `brokeredMsg` parameter of `GetMessageFilter` when explicitly retrievi
 ## Upcoming Features
 
 - [ ] queue whitelisting
-- [ ] request and response global filter
 - [ ] error handler

src/ServiceStack.AzureServiceBus/AzureBusServer.cs

@@ -72,6 +72,18 @@ public Action<string, BrokeredMessage, IMessage> PublishMessageFilter
             set => messageFactory.PublishMessageFilter = value;
         }
 
+        /// <summary>
+        /// Execute global transformation or custom logic before a request is processed.
+        /// Must be thread-safe.
+        /// </summary>
+        public Func<IMessage, IMessage> RequestFilter { get; set; }
+
+        /// <summary>
+        /// Execute global transformation or custom logic on the response.
+        /// Must be thread-safe.
+        /// </summary>
+        public Func<object, object> ResponseFilter { get; set; }
+
         private int status;
 
         public AzureBusServer(string connectionString): this(new AzureBusMessageFactory(connectionString))
@@ -142,6 +154,8 @@ protected IMessageHandlerFactory CreateMessageHandlerFactory<T>(Func<IMessage<T>
         {
             return new MessageHandlerFactory<T>(this, processMessageFn, processExceptionEx)
             {
+                RequestFilter = RequestFilter,
+                ResponseFilter = ResponseFilter,
                 RetryCount = RetryCount,
             };
         }

tests/ServiceStack.AzureServiceBus.Tests/AzureBusServerTests.cs

@@ -155,10 +155,16 @@ public class Hello : IReturn<HelloResponse>
         {
             public string Name { get; set; }
         }
-        public class HelloNull : IReturn<HelloResponse>
+        public class HelloNull1 : IReturn<HelloResponse>
         {
             public string Name { get; set; }
         }
+
+        public class HelloNull2 : IReturn<HelloResponse>
+        {
+            public string Name { get; set; }
+        }
+
         public class HelloResponse
         {
             public string Result { get; set; }
@@ -307,8 +313,11 @@ public async Task Messages_with_null_Response_is_published_to_OutMQ()
             int msgsReceived = 0;
             using (var mqServer = CreateMqServer())
             {
-                await mqServer.MessageFactory.PurgeQueueAsync<HelloNull>();
-                mqServer.RegisterHandler<HelloNull>(m =>
+                await mqServer.MessageFactory.PurgeQueuesAsync(
+                    QueueNames<HelloNull1>.In,
+                    QueueNames<HelloNull1>.Out
+                    );
+                mqServer.RegisterHandler<HelloNull1>(m =>
                 {
                     Interlocked.Increment(ref msgsReceived);
                     return null;
@@ -318,12 +327,12 @@ public async Task Messages_with_null_Response_is_published_to_OutMQ()
 
                 using (var mqClient = mqServer.CreateMessageQueueClient())
                 {
-                    mqClient.Publish(new HelloNull { Name = "Into the Void" });
+                    mqClient.Publish(new HelloNull1 { Name = "Into the Void" });
 
-                    var msg = mqClient.Get<HelloNull>(QueueNames<HelloNull>.Out, TimeSpan.FromSeconds(10));
+                    var msg = mqClient.Get<HelloNull1>(QueueNames<HelloNull1>.Out, TimeSpan.FromSeconds(10));
                     Assert.That(msg, Is.Not.Null);
 
-                    HelloNull response = msg.GetBody();
+                    HelloNull1 response = msg.GetBody();
 
                     Thread.Sleep(100);
 
@@ -339,8 +348,11 @@ public async Task Messages_with_null_Response_is_published_to_ReplyMQ()
             int msgsReceived = 0;
             using (var mqServer = CreateMqServer())
             {
-                await mqServer.MessageFactory.PurgeQueuesAsync(QueueNames<HelloNull>.In);
-                mqServer.RegisterHandler<HelloNull>(m =>
+                await mqServer.MessageFactory.PurgeQueuesAsync(
+                   QueueNames<HelloNull2>.In,
+                   QueueNames<HelloNull2>.Out
+                   );
+                mqServer.RegisterHandler<HelloNull2>(m =>
                 {
                     Interlocked.Increment(ref msgsReceived);
                     return null;
@@ -351,16 +363,16 @@ public async Task Messages_with_null_Response_is_published_to_ReplyMQ()
                 using (var mqClient = mqServer.CreateMessageQueueClient())
                 {
                     var replyMq = mqClient.GetTempQueueName();
-                    mqClient.Publish(new Message<HelloNull>(new HelloNull { Name = "Into the Void" })
+                    mqClient.Publish(new Message<HelloNull2>(new HelloNull2 { Name = "Into the Void" })
                     {
                         ReplyTo = replyMq
                     });
 
-                    var msg = mqClient.Get<HelloNull>(replyMq, TimeSpan.FromSeconds(10));
+                    var msg = mqClient.Get<HelloNull2>(replyMq, TimeSpan.FromSeconds(10));
 
-                    await Task.Delay(200);
+                    await Task.Delay(100);
 
-                    HelloNull response = msg.GetBody();
+                    HelloNull2 response = msg.GetBody();
                     Assert.That(response.Name, Is.EqualTo("Into the Void"));
                     Assert.That(msgsReceived, Is.EqualTo(1));
                 }

tests/ServiceStack.AzureServiceBus.Tests/MqServerIntroTests.cs

@@ -417,5 +417,46 @@ await Task.WhenAll(
                 appHost.Resolve<IMessageService>().Dispose();
             }
         }
+
+        [Test]
+        public async Task Global_request_and_response_get_called()
+        {
+            using (var mqServer = CreateMqServer())
+            {
+                await Task.WhenAll(
+                    mqServer.MessageFactory.PurgeQueueAsync<HelloIntro>(),
+                    mqServer.MessageFactory.PurgeQueuesAsync(QueueNames<HelloIntroResponse>.In));
+
+                var azureServer = mqServer as AzureBusServer;
+                azureServer.RequestFilter = msg =>
+                {
+                    msg.Meta["prefix"] = "Global";
+                    return msg;
+                };
+
+                azureServer.ResponseFilter = response =>
+                {
+                    if (response is HelloIntroResponse helloResp)
+                    {
+                        helloResp.Result += "!!!";
+                    }
+
+                    return response;
+                };
+
+                mqServer.RegisterHandler<HelloIntro>(m =>
+                    new HelloIntroResponse { Result = $"Hello, {m.Meta["prefix"]} {m.GetBody().Name}".Fmt() });
+                mqServer.Start();
+
+                using (var mqClient = mqServer.CreateMessageQueueClient())
+                {
+                    mqClient.Publish(new HelloIntro { Name = "World" });
+
+                    IMessage<HelloIntroResponse> responseMsg = mqClient.Get<HelloIntroResponse>(QueueNames<HelloIntroResponse>.In);
+                    mqClient.Ack(responseMsg);
+                    Assert.That(responseMsg.GetBody().Result, Is.EqualTo("Hello, Global World!!!"));
+                }
+            }
+        }
     }
 }