RabbitMqPortal.cs

Go to the documentation of this file.

//-----------------------------------------------------------------------
// <copyright file="RabbitMqPortal.cs" company="Marimer LLC">
//     Copyright (c) Marimer LLC. All rights reserved.
//     Website: https://cslanet.com
// </copyright>
// <summary>Exposes server-side DataPortal functionality through RabbitMQ</summary>
//-----------------------------------------------------------------------
using System;
using System.Security.Principal;
using System.Threading.Tasks;
using Csla.Core;
using Csla.Serialization;
using Csla.Serialization.Mobile;
using Csla.Server;
using Csla.Server.Hosts.DataPortalChannel;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
 
namespace Csla.Channels.RabbitMq
{
  public class RabbitMqPortal : IDisposable
  {
    public string DataPortalUrl { get; private set; }
 
    private IConnection Connection;
    private IModel Channel;
    private string DataPortalQueueName;
 
    public int Timeout { get; set; } = 30;
 
    public RabbitMqPortal()
    {
      DataPortalUrl = ApplicationContext.DataPortalUrlString;
    }
 
    public RabbitMqPortal(string dataPortalUrl)
    {
      DataPortalUrl = dataPortalUrl;
    }
 
    private Uri DataPortalUri { get; set; }
 
    private void InitializeRabbitMQ()
    {
      if (Connection == null)
      {
        Console.WriteLine($"Initializing connection to {DataPortalUrl}");
        DataPortalUri = new Uri(DataPortalUrl);
        var url = DataPortalUri;
        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");
 
        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();
      }
    }
 
    public void StartListening()
    {
      InitializeRabbitMQ();
      Channel.QueueDeclare(
        queue: DataPortalQueueName,
        durable: false,
        exclusive: false,
        autoDelete: false,
        arguments: null);
 
      var consumer = new EventingBasicConsumer(Channel);
      consumer.Received += (model, ea) =>
      {
        Console.WriteLine($"Received {ea.BasicProperties.Type} for {ea.BasicProperties.CorrelationId} from {ea.BasicProperties.ReplyTo}");
        InvokePortal(ea, ea.Body.ToArray());
      };
      Console.WriteLine($"Listening on queue {DataPortalQueueName}");
      Channel.BasicConsume(queue: DataPortalQueueName, autoAck: true, consumer: consumer);
    }
 
    private async void InvokePortal(BasicDeliverEventArgs ea, byte[] requestData)
    {
      var result = new DataPortalResponse();
      try
      {
        var request = SerializationFormatterFactory.GetFormatter().Deserialize(requestData);
        result = await CallPortal(ea.BasicProperties.Type, request);
      }
      catch (Exception ex)
      {
        result.ErrorData = new DataPortalErrorInfo(ex);
      }
 
      try
      {
        var response = SerializationFormatterFactory.GetFormatter().Serialize(result);
        SendMessage(ea.BasicProperties.ReplyTo, ea.BasicProperties.CorrelationId, response);
      }
      catch (Exception ex)
      {
        try
        {
          result = new DataPortalResponse { ErrorData = new DataPortalErrorInfo(ex) };
          var response = SerializationFormatterFactory.GetFormatter().Serialize(result);
          SendMessage(ea.BasicProperties.ReplyTo, ea.BasicProperties.CorrelationId, response);
        }
        catch (Exception ex1)
        {
          Console.Error.WriteLine($"## ERROR {ex1.Message}");
        }
      }
    }
 
    private void SendMessage(string target, string correlationId, byte[] request)
    {
      InitializeRabbitMQ();
      var props = Channel.CreateBasicProperties();
      props.CorrelationId = correlationId;
      Channel.BasicPublish(
        exchange: "",
        routingKey: target,
        basicProperties: props,
        body: request);
    }
 
    private async Task<DataPortalResponse> CallPortal(string operation, object request)
    {
      DataPortalResponse result;
      switch (operation)
      {
        case "create":
          result = await Create((CriteriaRequest)request).ConfigureAwait(false);
          break;
 
        case "fetch":
          result = await Fetch((CriteriaRequest)request).ConfigureAwait(false);
          break;
 
        case "update":
          result = await Update((UpdateRequest)request).ConfigureAwait(false);
          break;
 
        case "delete":
          result = await Delete((CriteriaRequest)request).ConfigureAwait(false);
          break;
 
        default:
          throw new InvalidOperationException(operation);
      }
      return result;
    }
 
    public async Task<DataPortalResponse> Create(CriteriaRequest request)
    {
      var result = new DataPortalResponse();
      try
      {
        request = ConvertRequest(request);
 
        // unpack criteria data into object
        object criteria = GetCriteria(request.CriteriaData);
        if (criteria is Csla.DataPortalClient.PrimitiveCriteria)
        {
          criteria = ((Csla.DataPortalClient.PrimitiveCriteria)criteria).Value;
        }
 
        var objectType = Csla.Reflection.MethodCaller.GetType(AssemblyNameTranslator.GetAssemblyQualifiedName(request.TypeName), true);
        var context = new DataPortalContext(
          (IPrincipal)SerializationFormatterFactory.GetFormatter().Deserialize(request.Principal),
          true,
          request.ClientCulture,
          request.ClientUICulture,
          (ContextDictionary)SerializationFormatterFactory.GetFormatter().Deserialize(request.ClientContext),
          (ContextDictionary)SerializationFormatterFactory.GetFormatter().Deserialize(request.GlobalContext));
 
        var prtl = new Csla.Server.DataPortal();
        var dpr = await prtl.Create(objectType, criteria, context, true);
 
        if (dpr.Error != null)
          result.ErrorData = new DataPortalErrorInfo(dpr.Error);
        result.GlobalContext = SerializationFormatterFactory.GetFormatter().Serialize(dpr.GlobalContext);
        result.ObjectData = SerializationFormatterFactory.GetFormatter().Serialize(dpr.ReturnObject);
      }
      catch (Exception ex)
      {
        result.ErrorData = new DataPortalErrorInfo(ex);
        throw;
      }
      finally
      {
        result = ConvertResponse(result);
      }
      return result;
    }
 
    public async Task<DataPortalResponse> Fetch(CriteriaRequest request)
    {
      var result = new DataPortalResponse();
      try
      {
        request = ConvertRequest(request);
 
        // unpack criteria data into object
        object criteria = GetCriteria(request.CriteriaData);
        if (criteria is Csla.DataPortalClient.PrimitiveCriteria)
        {
          criteria = ((Csla.DataPortalClient.PrimitiveCriteria)criteria).Value;
        }
 
        var objectType = Csla.Reflection.MethodCaller.GetType(AssemblyNameTranslator.GetAssemblyQualifiedName(request.TypeName), true);
        var context = new DataPortalContext(
          (IPrincipal)SerializationFormatterFactory.GetFormatter().Deserialize(request.Principal),
          true,
          request.ClientCulture,
          request.ClientUICulture,
          (ContextDictionary)SerializationFormatterFactory.GetFormatter().Deserialize(request.ClientContext),
          (ContextDictionary)SerializationFormatterFactory.GetFormatter().Deserialize(request.GlobalContext));
 
        var prtl = new Csla.Server.DataPortal();
        var dpr = await prtl.Fetch(objectType, criteria, context, true);
 
        if (dpr.Error != null)
          result.ErrorData = new DataPortalErrorInfo(dpr.Error);
        result.GlobalContext = SerializationFormatterFactory.GetFormatter().Serialize(dpr.GlobalContext);
        result.ObjectData = SerializationFormatterFactory.GetFormatter().Serialize(dpr.ReturnObject);
      }
      catch (Exception ex)
      {
        result.ErrorData = new DataPortalErrorInfo(ex);
        throw;
      }
      finally
      {
        result = ConvertResponse(result);
      }
      return result;
    }
 
    public async Task<DataPortalResponse> Update(UpdateRequest request)
    {
      var result = new DataPortalResponse();
      try
      {
        request = ConvertRequest(request);
        // unpack object
        object obj = GetCriteria(request.ObjectData);
 
        var context = new DataPortalContext(
          (IPrincipal)SerializationFormatterFactory.GetFormatter().Deserialize(request.Principal),
          true,
          request.ClientCulture,
          request.ClientUICulture,
          (ContextDictionary)SerializationFormatterFactory.GetFormatter().Deserialize(request.ClientContext),
          (ContextDictionary)SerializationFormatterFactory.GetFormatter().Deserialize(request.GlobalContext));
 
        var prtl = new Csla.Server.DataPortal();
        var dpr = await prtl.Update(obj, context, true);
 
        if (dpr.Error != null)
          result.ErrorData = new DataPortalErrorInfo(dpr.Error);
 
        result.GlobalContext = SerializationFormatterFactory.GetFormatter().Serialize(dpr.GlobalContext);
        result.ObjectData = SerializationFormatterFactory.GetFormatter().Serialize(dpr.ReturnObject);
      }
      catch (Exception ex)
      {
        result.ErrorData = new DataPortalErrorInfo(ex);
        throw;
      }
      finally
      {
        result = ConvertResponse(result);
      }
      return result;
    }
 
    public async Task<DataPortalResponse> Delete(CriteriaRequest request)
    {
      var result = new DataPortalResponse();
      try
      {
        request = ConvertRequest(request);
 
        // unpack criteria data into object
        object criteria = GetCriteria(request.CriteriaData);
        if (criteria is Csla.DataPortalClient.PrimitiveCriteria)
        {
          criteria = ((Csla.DataPortalClient.PrimitiveCriteria)criteria).Value;
        }
 
        var objectType = Csla.Reflection.MethodCaller.GetType(AssemblyNameTranslator.GetAssemblyQualifiedName(request.TypeName), true);
        var context = new DataPortalContext(
          (IPrincipal)SerializationFormatterFactory.GetFormatter().Deserialize(request.Principal),
          true,
          request.ClientCulture,
          request.ClientUICulture,
          (ContextDictionary)SerializationFormatterFactory.GetFormatter().Deserialize(request.ClientContext),
          (ContextDictionary)SerializationFormatterFactory.GetFormatter().Deserialize(request.GlobalContext));
 
        var prtl = new Csla.Server.DataPortal();
        var dpr = await prtl.Delete(objectType, criteria, context, true);
 
        if (dpr.Error != null)
          result.ErrorData = new DataPortalErrorInfo(dpr.Error);
        result.GlobalContext = SerializationFormatterFactory.GetFormatter().Serialize(dpr.GlobalContext);
        result.ObjectData = SerializationFormatterFactory.GetFormatter().Serialize(dpr.ReturnObject);
      }
      catch (Exception ex)
      {
        result.ErrorData = new DataPortalErrorInfo(ex);
        throw;
      }
      finally
      {
        result = ConvertResponse(result);
      }
      return result;
    }
 
    #region Criteria
 
    private static object GetCriteria(byte[] criteriaData)
    {
      object criteria = null;
      if (criteriaData != null)
        criteria = SerializationFormatterFactory.GetFormatter().Deserialize(criteriaData);
      return criteria;
    }
 
    #endregion Criteria
 
    #region Conversion methods
 
    protected virtual UpdateRequest ConvertRequest(UpdateRequest request)
    {
      return request;
    }
 
    protected virtual CriteriaRequest ConvertRequest(CriteriaRequest request)
    {
      return request;
    }
 
    protected virtual DataPortalResponse ConvertResponse(DataPortalResponse response)
    {
      return response;
    }
 
    #endregion Conversion methods
 
    public void Dispose()
    {
      Connection?.Close();
      Channel?.Dispose();
      Connection?.Dispose();
    }
  }
}