RabbitMqProxy.cs

Go to the documentation of this file.

//-----------------------------------------------------------------------
// <copyright file="RabbitMqProxy.cs" company="Marimer LLC">
//     Copyright (c) Marimer LLC. All rights reserved.
//     Website: https://cslanet.com
// </copyright>
// <summary>Implements a data portal proxy to relay data portal</summary>
//-----------------------------------------------------------------------
using System;
using System.Threading.Tasks;
using Csla.DataPortalClient;
using Csla.Server;
using RabbitMQ.Client;
 
namespace Csla.Channels.RabbitMq
{
  public class RabbitMqProxy : DataPortalProxy, IDisposable
  {
    public RabbitMqProxy(ApplicationContext applicationContext, RabbitMqProxyOptions options)
      : base(applicationContext)
    {
      DataPortalUrl = options.DataPortalUrl;
    }
 
    public override int Timeout { get; set; } = 30;
 
    protected IConnection Connection { get; set; }
 
    protected IModel Channel { get; set; }
 
    protected string DataPortalQueueName { get; set; }
 
    private ProxyListener QueueListener { get; set; }
 
    protected virtual void InitializeRabbitMQ()
    {
      if (Connection == null)
      {
        var url = new Uri(DataPortalUrl);
        Console.WriteLine($"Initializing {DataPortalUrl}");
        if (url.Scheme != "rabbitmq")
          throw new UriFormatException("Scheme != rabbitmq://");
        if (string.IsNullOrWhiteSpace(url.Host))
          throw new UriFormatException("Host");
        DataPortalQueueName = url.AbsolutePath.Substring(1);
        if (string.IsNullOrWhiteSpace(DataPortalQueueName))
          throw new UriFormatException("DataPortalQueueName");
        Console.WriteLine($"Will send to queue {DataPortalQueueName}");
        var factory = new ConnectionFactory() { HostName = url.Host };
        if (url.Port < 0)
          factory.Port = url.Port;
        var userInfo = url.UserInfo.Split(':');
        if (userInfo.Length > 0 && !string.IsNullOrWhiteSpace(userInfo[0]))
          factory.UserName = userInfo[0];
        if (userInfo.Length > 1)
          factory.Password = userInfo[1];
        Connection = factory.CreateConnection();
        Channel = Connection.CreateModel();
        if (QueueListener == null)
        {
          QueueListener = ProxyListener.GetListener(url);
          QueueListener.StartListening();
        }
      }
    }
 
    public override async Task<DataPortalResult> Create(Type objectType, object criteria, DataPortalContext context, bool isSync)
    {
      if (isSync)
        throw new NotSupportedException("isSync == true");
 
      InitializeRabbitMQ();
      return await base.Create(objectType, criteria, context, isSync);
    }
 
    public override async Task<DataPortalResult> Fetch(Type objectType, object criteria, DataPortalContext context, bool isSync)
    {
      if (isSync)
        throw new NotSupportedException("isSync == true");
 
      InitializeRabbitMQ();
      return await base.Fetch(objectType, criteria, context, isSync);
    }
 
    public override async Task<DataPortalResult> Update(object obj, DataPortalContext context, bool isSync)
    {
      if (isSync)
        throw new NotSupportedException("isSync == true");
 
      InitializeRabbitMQ();
      return await base.Update(obj, context, isSync);
    }
 
    public override async Task<DataPortalResult> Delete(Type objectType, object criteria, DataPortalContext context, bool isSync)
    {
      if (isSync)
        throw new NotSupportedException("isSync == true");
 
      InitializeRabbitMQ();
      return await base.Delete(objectType, criteria, context, isSync);
    }
 
    protected override async Task<byte[]> CallDataPortalServer(byte[] serialized, string operation, string routingToken, bool isSync)
    {
      var correlationId = Guid.NewGuid().ToString();
      var resetEvent = new Csla.Threading.AsyncManualResetEvent();
      var wip = Wip.WorkInProgress.GetOrAdd(correlationId, new WipItem { ResetEvent = resetEvent });
 
      SendMessage(QueueListener.ReplyQueue.QueueName, correlationId, operation, serialized);
 
      var timeout = Task.Delay(Timeout * 1000);
      if (await Task.WhenAny(wip.ResetEvent.WaitAsync(), timeout) == timeout)
        throw new TimeoutException();
 
      return wip.Response;
    }
 
    private void SendMessage(string sender, string correlationId, string operation, byte[] request)
    {
      var props = Channel.CreateBasicProperties();
      if (!string.IsNullOrWhiteSpace(sender))
        props.ReplyTo = sender;
      props.CorrelationId = correlationId;
      props.Type = operation;
      Channel.BasicPublish(
        exchange: "",
        routingKey: DataPortalQueueName,
        basicProperties: props,
        body: request);
    }
 
    public void Dispose()
    {
      QueueListener?.Dispose();
      Connection?.Close();
      Channel?.Dispose();
      Connection?.Dispose();
      Channel = null;
      Connection = null;
    }
  }
}