echo/rabbitmq-tutorials
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

More consistent rpc client, more wrapping

Browse Source
This commit is contained in:
Emile Joubert 2011-06-13 15:46:58 +01:00
parent b47b345937
commit f9b06ed777
6 changed files with 58 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
dotnet/EmitLogDirect.cs
View File

@ -11,7 +11,8 @@ public static void Main(string[] args) {
channel.ExchangeDeclare("direct_logs", "direct");
string severity = (args.Length > 0) ? args[0] : "info";
string message = (args.Length > 1) ? string.Join(" ", args.Skip(1).ToArray())
string message = (args.Length > 1) ? string.Join(" ", args.Skip(1)
.ToArray())
: "Hello World!";
byte[] body = System.Text.Encoding.UTF8.GetBytes(message);
channel.BasicPublish("direct_logs", severity, null, body);

3
dotnet/EmitLogTopic.cs
View File

@ -11,7 +11,8 @@ public static void Main(string[] args) {
channel.ExchangeDeclare("topic_logs", "topic");
string routingKey = (args.Length > 0) ? args[0] : "anonymous.info";
string message = (args.Length > 1) ? string.Join(" ", args.Skip(1).ToArray())
string message = (args.Length > 1) ? string.Join(" ", args.Skip(1)
.ToArray())
: "Hello World!";
byte[] body = System.Text.Encoding.UTF8.GetBytes(message);
channel.BasicPublish("topic_logs", routingKey, null, body);

64
dotnet/RPCClient.cs
View File

@ -2,41 +2,53 @@
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
class RPCClient {
private static string RpcCall(string message) {
string response = null;
class RPCClient : IDisposable {
private IConnection connection;
private IModel channel;
private string replyQueueName;
private QueueingBasicConsumer consumer;
public RPCClient() {
ConnectionFactory factory = new ConnectionFactory();
factory.HostName = "localhost";
using (IConnection connection = factory.CreateConnection())
using (IModel channel = connection.CreateModel()) {
string replyQueueName = channel.QueueDeclare();
QueueingBasicConsumer consumer = new QueueingBasicConsumer(channel);
channel.BasicConsume(replyQueueName, false, consumer);
connection = factory.CreateConnection();
channel = connection.CreateModel();
replyQueueName = channel.QueueDeclare();
consumer = new QueueingBasicConsumer(channel);
channel.BasicConsume(replyQueueName, false, consumer);
}
string corrId = Guid.NewGuid().ToString();
IBasicProperties props = channel.CreateBasicProperties();
props.ReplyTo = replyQueueName;
props.CorrelationId = corrId;
public string Call(string message) {
string response = null;
string corrId = Guid.NewGuid().ToString();
IBasicProperties props = channel.CreateBasicProperties();
props.ReplyTo = replyQueueName;
props.CorrelationId = corrId;
byte[] messageBytes = System.Text.Encoding.UTF8.GetBytes(message);
channel.BasicPublish("", "rpc_queue", props, messageBytes);
byte[] messageBytes = System.Text.Encoding.UTF8.GetBytes(message);
channel.BasicPublish("", "rpc_queue", props, messageBytes);
while (true) {
BasicDeliverEventArgs ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
if (ea.BasicProperties.CorrelationId == corrId) {
byte[] body = ea.Body;
response = System.Text.Encoding.UTF8.GetString(body);
channel.BasicCancel(consumer.ConsumerTag);
break;
}
while (true) {
BasicDeliverEventArgs ea =
(BasicDeliverEventArgs)consumer.Queue.Dequeue();
if (ea.BasicProperties.CorrelationId == corrId) {
byte[] body = ea.Body;
response = System.Text.Encoding.UTF8.GetString(body);
channel.BasicCancel(consumer.ConsumerTag);
break;
}
return response;
}
return response;
}
public void Dispose() {
connection.Close();
}
public static void Main() {
Console.WriteLine(" [x] Requesting fib(30)");
string response = RpcCall("30");
Console.WriteLine(" [.] Got '{0}'", response);
using (RPCClient rpcClient = new RPCClient()) {
string response = rpcClient.Call("30");
Console.WriteLine(" [.] Got '{0}'", response);
}
}
}
}

9
dotnet/RPCServer.cs
View File

@ -16,7 +16,8 @@ public static void Main() {
while(true) {
string response = null;
BasicDeliverEventArgs ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
BasicDeliverEventArgs ea =
(BasicDeliverEventArgs)consumer.Queue.Dequeue();
byte[] body = ea.Body;
IBasicProperties props = ea.BasicProperties;
@ -32,8 +33,10 @@ public static void Main() {
Console.WriteLine(" [.] " + e);
response = "";
} finally {
byte[] responseBytes = System.Text.Encoding.UTF8.GetBytes(response);
channel.BasicPublish("", props.ReplyTo, replyProps, responseBytes);
byte[] responseBytes =
System.Text.Encoding.UTF8.GetBytes(response);
channel.BasicPublish("", props.ReplyTo, replyProps,
responseBytes);
channel.BasicAck(ea.DeliveryTag, false);
}
}

6
dotnet/ReceiveLogsDirect.cs
View File

@ -23,7 +23,8 @@ public static void Main(string[] args) {
channel.QueueBind(queue_name, "direct_logs", severity);
}
Console.WriteLine(" [*] Waiting for messages. To exit press CTRL+C");
Console.WriteLine(" [*] Waiting for messages. " +
"To exit press CTRL+C");
QueueingBasicConsumer consumer = new QueueingBasicConsumer(channel);
channel.BasicConsume(queue_name, true, consumer);
@ -35,7 +36,8 @@ public static void Main(string[] args) {
byte[] body = ea.Body;
string message = System.Text.Encoding.UTF8.GetString(body);
string routingKey = ea.RoutingKey;
Console.WriteLine(" [x] Received '{0}':'{1}'", routingKey, message);
Console.WriteLine(" [x] Received '{0}':'{1}'",
routingKey, message);
}
}
}

9
dotnet/ReceiveLogsTopic.cs
View File

@ -22,9 +22,11 @@ public static void Main(string[] args) {
channel.QueueBind(queue_name, "topic_logs", bindingKey);
}
Console.WriteLine(" [*] Waiting for messages. To exit press CTRL+C");
Console.WriteLine(" [*] Waiting for messages. " +
"To exit press CTRL+C");
QueueingBasicConsumer consumer = new QueueingBasicConsumer(channel);
QueueingBasicConsumer consumer =
new QueueingBasicConsumer(channel);
channel.BasicConsume(queue_name, true, consumer);
while(true) {
@ -34,7 +36,8 @@ public static void Main(string[] args) {
byte[] body = ea.Body;
string message = System.Text.Encoding.UTF8.GetString(body);
string routingKey = ea.RoutingKey;
Console.WriteLine(" [x] Received '{0}':'{1}'", routingKey, message);
Console.WriteLine(" [x] Received '{0}':'{1}'",
routingKey, message);
}
}
}