Skip to content
Permalink
master
Switch branches/tags
Go to file
 
 
Cannot retrieve contributors at this time
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Concurrent;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
public class RpcClient
{
private const string QUEUE_NAME = "rpc_queue";
private readonly IConnection connection;
private readonly IModel channel;
private readonly string replyQueueName;
private readonly EventingBasicConsumer consumer;
private readonly ConcurrentDictionary<string, TaskCompletionSource<string>> callbackMapper =
new ConcurrentDictionary<string, TaskCompletionSource<string>>();
public RpcClient()
{
var factory = new ConnectionFactory() { HostName = "localhost" };
connection = factory.CreateConnection();
channel = connection.CreateModel();
// declare a server-named queue
replyQueueName = channel.QueueDeclare(queue: "").QueueName;
consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
if (!callbackMapper.TryRemove(ea.BasicProperties.CorrelationId, out TaskCompletionSource<string> tcs))
return;
var body = ea.Body.ToArray();
var response = Encoding.UTF8.GetString(body);
tcs.TrySetResult(response);
};
channel.BasicConsume(
consumer: consumer,
queue: replyQueueName,
autoAck: true);
}
public Task<string> CallAsync(string message, CancellationToken cancellationToken = default(CancellationToken))
{
IBasicProperties props = channel.CreateBasicProperties();
var correlationId = Guid.NewGuid().ToString();
props.CorrelationId = correlationId;
props.ReplyTo = replyQueueName;
var messageBytes = Encoding.UTF8.GetBytes(message);
var tcs = new TaskCompletionSource<string>();
callbackMapper.TryAdd(correlationId, tcs);
channel.BasicPublish(
exchange: "",
routingKey: QUEUE_NAME,
basicProperties: props,
body: messageBytes);
cancellationToken.Register(() => callbackMapper.TryRemove(correlationId, out var tmp));
return tcs.Task;
}
public void Close()
{
channel.Close();
connection.Close();
}
}
public class Rpc
{
public static void Main(string[] args)
{
Console.WriteLine("RPC Client");
string n = args.Length > 0 ? args[0] : "30";
Task t = InvokeAsync(n);
t.Wait();
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
private static async Task InvokeAsync(string n)
{
var rpcClient = new RpcClient();
Console.WriteLine(" [x] Requesting fib({0})", n);
var response = await rpcClient.CallAsync(n.ToString());
Console.WriteLine(" [.] Got '{0}'", response);
rpcClient.Close();
}
}