Rewrite .NET RPC Server to use EventingBasicConsumer

olsh · olsh · commit 2049e4c28617 · 2016-02-11T17:10:29.000+03:00
diff --git a/dotnet-visual-studio/6_RPCServer/Program.cs b/dotnet-visual-studio/6_RPCServer/Program.cs
@@ -8,19 +8,18 @@ class Program
     public static void Main()
     {
         var factory = new ConnectionFactory() { HostName = "localhost" };
-        using(var connection = factory.CreateConnection())
-        using(var channel = connection.CreateModel())
+        using (var connection = factory.CreateConnection())
+        using (var channel = connection.CreateModel())
         {
             channel.QueueDeclare(queue: "rpc_queue", durable: false, exclusive: false, autoDelete: false, arguments: null);
             channel.BasicQos(0, 1, false);
-            var consumer = new QueueingBasicConsumer(channel);
+            var consumer = new EventingBasicConsumer(channel);
             channel.BasicConsume(queue: "rpc_queue", noAck: false, consumer: consumer);
             Console.WriteLine(" [x] Awaiting RPC requests");
 
-            while(true)
+            consumer.Received += (model, ea) =>
             {
                 string response = null;
-                var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
 
                 var body = ea.Body;
                 var props = ea.BasicProperties;
@@ -34,7 +33,7 @@ public static void Main()
                     Console.WriteLine(" [.] fib({0})", message);
                     response = fib(n).ToString();
                 }
-                catch(Exception e)
+                catch (Exception e)
                 {
                     Console.WriteLine(" [.] " + e.Message);
                     response = "";
@@ -45,7 +44,10 @@ public static void Main()
                     channel.BasicPublish(exchange: "", routingKey: props.ReplyTo, basicProperties: replyProps, body: responseBytes);
                     channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                 }
-            }
+            };
+
+            Console.WriteLine(" Press [enter] to exit.");
+            Console.ReadLine();
         }
     }
 
@@ -55,7 +57,7 @@ public static void Main()
     /// </summary>
     private static int fib(int n)
     {
-        if(n == 0 || n == 1)
+        if (n == 0 || n == 1)
         {
             return n;
         }
diff --git a/dotnet/RPCServer.cs b/dotnet/RPCServer.cs
@@ -8,19 +8,18 @@ class RPCServer
     public static void Main()
     {
         var factory = new ConnectionFactory() { HostName = "localhost" };
-        using(var connection = factory.CreateConnection())
-        using(var channel = connection.CreateModel())
+        using (var connection = factory.CreateConnection())
+        using (var channel = connection.CreateModel())
         {
             channel.QueueDeclare(queue: "rpc_queue", durable: false, exclusive: false, autoDelete: false, arguments: null);
             channel.BasicQos(0, 1, false);
-            var consumer = new QueueingBasicConsumer(channel);
+            var consumer = new EventingBasicConsumer(channel);
             channel.BasicConsume(queue: "rpc_queue", noAck: false, consumer: consumer);
             Console.WriteLine(" [x] Awaiting RPC requests");
 
-            while(true)
+            consumer.Received += (model, ea) =>
             {
                 string response = null;
-                var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
 
                 var body = ea.Body;
                 var props = ea.BasicProperties;
@@ -34,7 +33,7 @@ public static void Main()
                     Console.WriteLine(" [.] fib({0})", message);
                     response = fib(n).ToString();
                 }
-                catch(Exception e)
+                catch (Exception e)
                 {
                     Console.WriteLine(" [.] " + e.Message);
                     response = "";
@@ -45,7 +44,10 @@ public static void Main()
                     channel.BasicPublish(exchange: "", routingKey: props.ReplyTo, basicProperties: replyProps, body: responseBytes);
                     channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                 }
-            }
+            };
+
+            Console.WriteLine(" Press [enter] to exit.");
+            Console.ReadLine();
         }
     }
 
@@ -55,7 +57,7 @@ public static void Main()
     /// </summary>
     private static int fib(int n)
     {
-        if(n == 0 || n == 1)
+        if (n == 0 || n == 1)
         {
             return n;
         }

Original file line number	Diff line number	Diff line change
`@@ -8,19 +8,18 @@ class Program`
`8`	`8`	`public static void Main()`
`9`	`9`	`{`
`10`	`10`	`var factory = new ConnectionFactory() { HostName = "localhost" };`
`11`		`- using(var connection = factory.CreateConnection())`
`12`		`- using(var channel = connection.CreateModel())`
	`11`	`+ using (var connection = factory.CreateConnection())`
	`12`	`+ using (var channel = connection.CreateModel())`
`13`	`13`	`{`
`14`	`14`	`channel.QueueDeclare(queue: "rpc_queue", durable: false, exclusive: false, autoDelete: false, arguments: null);`
`15`	`15`	`channel.BasicQos(0, 1, false);`
`16`		`- var consumer = new QueueingBasicConsumer(channel);`
	`16`	`+ var consumer = new EventingBasicConsumer(channel);`
`17`	`17`	`channel.BasicConsume(queue: "rpc_queue", noAck: false, consumer: consumer);`
`18`	`18`	`Console.WriteLine(" [x] Awaiting RPC requests");`
`19`	`19`
`20`		`- while(true)`
	`20`	`+ consumer.Received += (model, ea) =>`
`21`	`21`	`{`
`22`	`22`	`string response = null;`
`23`		`- var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();`
`24`	`23`
`25`	`24`	`var body = ea.Body;`
`26`	`25`	`var props = ea.BasicProperties;`
`@@ -34,7 +33,7 @@ public static void Main()`
`34`	`33`	`Console.WriteLine(" [.] fib({0})", message);`
`35`	`34`	`response = fib(n).ToString();`
`36`	`35`	`}`
`37`		`- catch(Exception e)`
	`36`	`+ catch (Exception e)`
`38`	`37`	`{`
`39`	`38`	`Console.WriteLine(" [.] " + e.Message);`
`40`	`39`	`response = "";`
`@@ -45,7 +44,10 @@ public static void Main()`
`45`	`44`	`channel.BasicPublish(exchange: "", routingKey: props.ReplyTo, basicProperties: replyProps, body: responseBytes);`
`46`	`45`	`channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);`
`47`	`46`	`}`
`48`		`- }`
	`47`	`+ };`
	`48`	`+`
	`49`	`+ Console.WriteLine(" Press [enter] to exit.");`
	`50`	`+ Console.ReadLine();`
`49`	`51`	`}`
`50`	`52`	`}`
`51`	`53`
`@@ -55,7 +57,7 @@ public static void Main()`
`55`	`57`	`/// </summary>`
`56`	`58`	`private static int fib(int n)`
`57`	`59`	`{`
`58`		`- if(n == 0 \|\| n == 1)`
	`60`	`+ if (n == 0 \|\| n == 1)`
`59`	`61`	`{`
`60`	`62`	`return n;`
`61`	`63`	`}`
Original file line number	Diff line number	Diff line change
`@@ -8,19 +8,18 @@ class RPCServer`
`8`	`8`	`public static void Main()`
`9`	`9`	`{`
`10`	`10`	`var factory = new ConnectionFactory() { HostName = "localhost" };`
`11`		`- using(var connection = factory.CreateConnection())`
`12`		`- using(var channel = connection.CreateModel())`
	`11`	`+ using (var connection = factory.CreateConnection())`
	`12`	`+ using (var channel = connection.CreateModel())`
`13`	`13`	`{`
`14`	`14`	`channel.QueueDeclare(queue: "rpc_queue", durable: false, exclusive: false, autoDelete: false, arguments: null);`
`15`	`15`	`channel.BasicQos(0, 1, false);`
`16`		`- var consumer = new QueueingBasicConsumer(channel);`
	`16`	`+ var consumer = new EventingBasicConsumer(channel);`
`17`	`17`	`channel.BasicConsume(queue: "rpc_queue", noAck: false, consumer: consumer);`
`18`	`18`	`Console.WriteLine(" [x] Awaiting RPC requests");`
`19`	`19`
`20`		`- while(true)`
	`20`	`+ consumer.Received += (model, ea) =>`
`21`	`21`	`{`
`22`	`22`	`string response = null;`
`23`		`- var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();`
`24`	`23`
`25`	`24`	`var body = ea.Body;`
`26`	`25`	`var props = ea.BasicProperties;`
`@@ -34,7 +33,7 @@ public static void Main()`
`34`	`33`	`Console.WriteLine(" [.] fib({0})", message);`
`35`	`34`	`response = fib(n).ToString();`
`36`	`35`	`}`
`37`		`- catch(Exception e)`
	`36`	`+ catch (Exception e)`
`38`	`37`	`{`
`39`	`38`	`Console.WriteLine(" [.] " + e.Message);`
`40`	`39`	`response = "";`
`@@ -45,7 +44,10 @@ public static void Main()`
`45`	`44`	`channel.BasicPublish(exchange: "", routingKey: props.ReplyTo, basicProperties: replyProps, body: responseBytes);`
`46`	`45`	`channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);`
`47`	`46`	`}`
`48`		`- }`
	`47`	`+ };`
	`48`	`+`
	`49`	`+ Console.WriteLine(" Press [enter] to exit.");`
	`50`	`+ Console.ReadLine();`
`49`	`51`	`}`
`50`	`52`	`}`
`51`	`53`
`@@ -55,7 +57,7 @@ public static void Main()`
`55`	`57`	`/// </summary>`
`56`	`58`	`private static int fib(int n)`
`57`	`59`	`{`
`58`		`- if(n == 0 \|\| n == 1)`
	`60`	`+ if (n == 0 \|\| n == 1)`
`59`	`61`	`{`
`60`	`62`	`return n;`
`61`	`63`	`}`