<commit_before>﻿//-----------------------------------------------------------------------
// <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.IO;
using System.Security.Principal;
using System.Threading.Tasks;
using Csla.Core;
using Csla.Serialization.Mobile;
using Csla.Server;
using Csla.Server.Hosts.HttpChannel;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace Csla.Channels.RabbitMq
{
  /// <summary>
  /// Exposes server-side DataPortal functionality through RabbitMQ
  /// </summary>
  public class RabbitMqPortal : IDisposable
  {
    /// <summary>
    /// Gets the URI for the data portal service.
    /// </summary>
    public string DataPortalUrl { get; private set; }
    private IConnection Connection;
    private IModel Channel;
    private string DataPortalQueueName;
    /// <summary>
    /// Gets or sets the timeout for network
    /// operations in seconds (default is 30 seconds).
    /// </summary>
    public int Timeout { get; set; } = 30;

    /// <summary>
    /// Creates an instance of the object.
    /// </summary>
    public RabbitMqPortal()
    {
      DataPortalUrl = ApplicationContext.DataPortalUrlString;
    }

    /// <summary>
    /// Creates an instance of the object.
    /// </summary>
    /// <param name="dataPortalUrl">URI for the data portal</param>
    public RabbitMqPortal(string dataPortalUrl)
    {
      DataPortalUrl = dataPortalUrl;
    }

    private Uri DataPortalUri { get; set; }

    private void InitializeRabbitMQ()
    {
      Console.WriteLine($"Initializing {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");
      if (Connection == null)
      {
        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();
        Connection.ConnectionShutdown += (s, e) =>
        {
          Connection?.Dispose();
          Connection = null;
          Channel = null;
        };
      }
      if (Channel == null)
      {
        Channel = Connection.CreateModel();
        Channel.ModelShutdown += (s, e) =>
        {
          Channel?.Dispose();
          Channel = null;
        };
      }
    }

    /// <summary>
    /// Start processing inbound messages.
    /// </summary>
    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);
      };
      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 HttpResponse();
      try
      {
        var request = MobileFormatter.Deserialize(requestData);
        result = await CallPortal(ea.BasicProperties.Type, request);
      }
      catch (Exception ex)
      {
        result.ErrorData = new HttpErrorInfo(ex);
      }

      try
      {
        var response = MobileFormatter.Serialize(result);
        SendMessage(ea.BasicProperties.ReplyTo, ea.BasicProperties.CorrelationId, response);
      }
      catch (Exception ex)
      {
        result = new HttpResponse { ErrorData = new HttpErrorInfo(ex) };
        var response = MobileFormatter.Serialize(result);
        SendMessage(ea.BasicProperties.ReplyTo, ea.BasicProperties.CorrelationId, response);
      }
    }

    private void SendMessage(string target, string correlationId, byte[] request)
    {
      var props = Channel.CreateBasicProperties();
      props.CorrelationId = correlationId;<commit_msg>#1066 Don't completely fail in service; improve retry functionality<commit_after>﻿//-----------------------------------------------------------------------
// <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.IO;
using System.Security.Principal;
using System.Threading.Tasks;
using Csla.Core;
using Csla.Serialization.Mobile;
using Csla.Server;
using Csla.Server.Hosts.HttpChannel;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace Csla.Channels.RabbitMq
{
  /// <summary>
  /// Exposes server-side DataPortal functionality through RabbitMQ
  /// </summary>
  public class RabbitMqPortal : IDisposable
  {
    /// <summary>
    /// Gets the URI for the data portal service.
    /// </summary>
    public string DataPortalUrl { get; private set; }
    private IConnection Connection;
    private IModel Channel;
    private string DataPortalQueueName;
    /// <summary>
    /// Gets or sets the timeout for network
    /// operations in seconds (default is 30 seconds).
    /// </summary>
    public int Timeout { get; set; } = 30;

    /// <summary>
    /// Creates an instance of the object.
    /// </summary>
    public RabbitMqPortal()
    {
      DataPortalUrl = ApplicationContext.DataPortalUrlString;
    }

    /// <summary>
    /// Creates an instance of the object.
    /// </summary>
    /// <param name="dataPortalUrl">URI for the data portal</param>
    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 = <PASSWORD>[1];
        Connection = factory.CreateConnection();
        Connection.ConnectionShutdown += (s, e) =>
        {
          Connection?.Dispose();
          Connection = null;
          Channel = null;
        };
      }
      if (Channel == null)
      {
        Channel = Connection.CreateModel();
        Channel.ModelShutdown += (s, e) =>
        {
          Channel?.Dispose();
          Channel = null;
        };
      }
    }

    /// <summary>
    /// Start processing inbound messages.
    /// </summary>
    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);
      };
      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 HttpResponse();
      try
      {
        var request = MobileFormatter.Deserialize(requestData);
        result = await CallPortal(ea.BasicProperties.Type, request);
      }
      catch (Exception ex)
      {
        result.ErrorData = new HttpErrorInfo(ex);
      }

      try
      {
        var response = MobileFormatter.Serialize(result);
        SendMessage(ea.BasicProperties.ReplyTo, ea.BasicProperties.CorrelationId, response);
      }
      catch (Exception ex)
      {
        try
        {
          result = new HttpResponse { ErrorData = new HttpErrorInfo(ex) };
          var response = MobileFormatter.Serialize(result);
          SendMessage(ea.BasicProperties.ReplyTo, ea.BasicProperties.CorrelationId, response);
        }
        catch (Exception ex1)
        {
          Console.WriteLine($"## ERROR {ex1.Message}");
        }
      }
    }

    private void SendMessage(string target, string correlationId, byte[] request)
    {
      InitializeRabbitMQ();
      var props = Channel.CreateBasicProperties();
      props.CorrelationId = correlationId;
{'fraction_non_alphanumeric': 0.13924175283111767, 'fraction_numerical': 0.00275726243229936, 'mean_word_length': 2.282482223658694, 'pattern_counts': {'":': 0, '<': 42, '<?xml version=': 0, '>': 50, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6711837', 'n_tokens_mistral': 2847, 'n_tokens_neox': 2767, 'n_words': 663}