using Edge.Core.Processor;using Edge.Core.IndustryStandardInterface.Pump;
using MessageRouter;
using Microsoft.Extensions.Logging;
using Edge.Core.Parser;
using System;
using System.Collections.Specialized;
using System.Linq;
using System.Threading;
namespace GlobalCommunicator
{
public class MessageRouterCommunicator<T> : IDisposable
{
public static ILogger fccLogger { get; set; }
private const string DEFAULT_FUSION_USER_NAME = "SSF"; // default fusion user id
private const string DEFAULT_FUSION_PASSWORD = "MINIMEINFINITERINGTONES"; // default fusion password
/// <summary>
/// 0 for not started, 1 for started already.
/// </summary>
private int isStarted = 0;
private IMessageParser<object, T> parser;
public event EventHandler<FccDataReceivedEventArgs> OnDataReceived;
private readonly MessageRouterClient msgRouterClient;
/// <summary>
/// Will automatically call the Start() for MessageRouterClient.
/// </summary>
/// <param name="msgRouterClient">unstarted MessageRouterClient</param>
public MessageRouterCommunicator(IMessageParser<object, T> parser, MessageRouterClient msgRouterClient)
{
this.parser = parser;
this.msgRouterClient = msgRouterClient;
this.msgRouterClient.OnConnect += (conn, reconnect) =>
{
fccLogger.LogDebug("Connected with MessageRouterServer");
string clientId = DEFAULT_FUSION_USER_NAME;
string password = DEFAULT_FUSION_PASSWORD;
string userName = clientId;
userName = MsgRouterMessageUtility.AppendSpaceOnRight(userName, 20);
password = MsgRouterMessageUtility.AppendSpaceOnRight(password, 25);
char[] cryptedPwd = MsgRouterMessageUtility.Crypt(password.ToCharArray(), 25, userName.ToCharArray(), 20);
password = MsgRouterMessageUtility.BinToHexString(cryptedPwd, 25);
if (!msgRouterClient.SendMessage(MsgRouterMessageUtility.Login(clientId.ToUpper(), password)))
fccLogger.LogDebug("Login failed to send to msgRouter server");
SendSubscriptions();
};
this.msgRouterClient.OnDisconnect += (client, exp) => fccLogger.LogDebug("!!!msgRouterClient disconnected with MessageRouterServer");
this.msgRouterClient.OnMessageRouterMessageReceived += this.OnMessageRouterMessageReceived;
}
private void OnMessageRouterMessageReceived(MessageRouterClient client, string msgType, string evtType, StringDictionary parameters)
{
fccLogger.LogDebug("Message received from MessageRouterServer");
fccLogger.LogDebug(" message type: " + msgType);
fccLogger.LogDebug(" event type: " + evtType);
fccLogger.LogDebug(" params(name->value): " + (parameters.Count == 0 ? "" : parameters.Keys.Cast<string>().Select(k => k + "->" + parameters[k]).Aggregate((acc, n) => acc + ", " + n)));
if (msgType == "ECHO")
{
parameters["PumpID"] = "-1";
parameters["EventType"] = "EchoEvent";
this.OnDataReceived?.Invoke(this, new FccDataReceivedEventArgs(parameters));
}
else if (msgType == "POST")
ProcessMessage(msgType, evtType, parameters);
}
private void ProcessMessage(string msgType, string evtType, StringDictionary parameters)
{
string evt = "";
string pumpId = "";
if (evtType.IndexOf("_ID_") > 0)
{
evt = evtType.Substring(0, evtType.Length - 7);
pumpId = evtType.Substring(evtType.Length - 3, 3);
}
else
{
evt = evtType;
}
switch (evt)
{
#region
case "RES_GENERIC_ERROR":
break;
case "RES_FCRT_PUMPS_CONFIG":
break;
case "EVT_PUMP_STATUS_CHANGE":
ProcessPumpStatusChange(parameters, Convert.ToInt32(pumpId));
break;
case "EVT_PUMP_DELIVERY_PROGRESS":
ProcessPumpDeliveryProgressChange(parameters, Convert.ToInt32(pumpId));
break;
case "RES_SECU_LOGIN":
break;
case "RES_SECU_ACCESS_DENIED":
break;
case "RES_PRICES_REFRESH_PRICE_CHANGE_TBL":
//LoadNewPrices();
break;
case "RES_PRICES_SET_NEW_PRICE_CHANGE":
//RefreshPriceChangeTable();
break;
case "EVT_NEW_PRICE_CHANGE_APPLIED":
//SendFuelPriceChange();
break;
case "REQ_RECEIVE_PRESET_FROM_FORECOURT":
//ProcessPresetFromFPOS(parameters, Convert.ToInt32(pumpId));
break;
case "REQ_RECEIVE_OUTDOOR_TRANSACTION_DENIED":
//ProcessOutdoorTrxDeniedFromFPOS(parameters, Convert.ToInt32(pumpId));
break;
case "REQ_RECEIVE_OUTDOOR_TRANSACTION_APPROVED":
//ProcessOutdoorTrxApprovedFromFPOS(parameters, Convert.ToInt32(pumpId));
break;
case "REQ_RECEIVE_OUTDOOR_AIRLOCK":
//ProcessOutdoorTrxAirlock(parameters, Convert.ToInt32(pumpId));
break;
default:
break;
#endregion
}
}
private void ProcessPumpDeliveryProgressChange(StringDictionary parameters, int pumpId)
{
if (string.IsNullOrEmpty(parameters["GR"]) || string.IsNullOrEmpty(parameters["AM"]) ||
string.IsNullOrEmpty(parameters["PU"]) || string.IsNullOrEmpty(parameters["VO"]))
{
fccLogger.LogDebug("Ignore ProcessPumpDeliveryProgressChange event for pump: " + pumpId);
return;
}
try
{
//parameters["PumpID"] = pumpId.ToString();
//parameters["EventType"] = "FuellingStatusChange";
//parameters["Finished"] = "false";
//fccLogger.LogDebug("ProcessPumpDeliveryProgressChange event for pump: " + pumpId);
//foreach (string key in parameters.Keys)
//{
// fccLogger.LogDebug(string.Format("Key: {0}, Value: {1}", key, parameters[key]));
//}
//this.OnDataReceived?.Invoke(this, new FccDataReceivedEventArgs(parameters));
//var gradeFriendlyName = Translator.GetFriendlyGradeName(SiteConfigUtility.Default.GetGradeNameByGradeId(int.Parse(gradeId)));
//using (var posSqlConnection = new SqlConnection(ConfigurationManager.ConnectionStrings["PosDatabaseConnStr"].ConnectionString))
//{
// try
// {
// var updatePumpOnFuelingCommand =
// new SqlCommand(string.Format("Update jy_info set [status] = '{1}', youpin = N'{2}', qty= {3}, amount= {4} where jihao = '{0}'",
// SiteConfigUtility.Default.GetSiteLevelNozzleIdByLogicalNozzleId(p, logicalHoseId),
// 'B',
// gradeFriendlyName,
// currentQty,
// currentAmount),
// posSqlConnection);
// fccLogger.LogDebug("updatePumpOnFuelingCommand: " + updatePumpOnFuelingCommand.CommandText, DebugLogLevel.Maximized);
// posSqlConnection.Open();
// updatePumpOnFuelingCommand.ExecuteNonQuery();
// }
// catch (Exception ex)
// {
// fccLogger.LogDebug("executing updatePumpOnFuelingCommand failed, exception detail: " + ex,
// DebugLogLevel.Normal);
// }
//}
}
catch (Exception ex)
{
fccLogger.LogDebug("Exception in handling ProcessPumpDeliveryProgressChange:" + ex);
}
}
private void ProcessPumpStatusChange(StringDictionary parameters, int pumpId)
{
try
{
parameters["PumpID"] = pumpId.ToString();
parameters["EventType"] = "PumpStatusChange";
fccLogger.LogDebug("ProcessPumpStatusChange event for pump: " + pumpId);
foreach (string key in parameters.Keys)
{
fccLogger.LogDebug(string.Format("Key: {0}, Value: {1}", key, parameters[key]));
}
this.OnDataReceived?.Invoke(this, new FccDataReceivedEventArgs(parameters));
}
catch (Exception ex)
{
fccLogger.LogError(" Exception in ProcessPumpStatusChange: " + parameters.ToString() + ", detail: \r\n" + ex.ToString());
}
//if (pumpStatusStr == "AUTHORIZED")
//{
// /* indicate a nozzle was lifted, and pending for start fueling */
//}
//else if (pumpStatusStr == "STARTING")
//{
// /* I understood this state as an intermidea, very short period once the trigger was clicked on nozzle, can be treated as indicator for fueling is ongoing */
// var gradeId = parameters["GR"];
// var hoseId = int.Parse(parameters["ho"]);
// var gradeFriendlyName = Translator.GetFriendlyGradeName(SiteConfigUtility.Default.GetGradeNameByGradeId(int.Parse(gradeId)));
// using (var posSqlConnection = new SqlConnection(ConfigurationManager.ConnectionStrings["PosDatabaseConnStr"].ConnectionString))
// {
// try
// {
// var setPumpOnStartingCommand =
// new SqlCommand(string.Format(
// "Update jy_info set [status] = '{1}', youpin = N'{2}', qty= 0, amount= 0 where jihao = '{0}'"
// , SiteConfigUtility.Default.GetSiteLevelNozzleIdByLogicalNozzleId(p, hoseId), 'B', gradeFriendlyName), posSqlConnection);
// fccLogger.LogDebug("setPumpOnStartingCommand: " + setPumpOnStartingCommand.CommandText, DebugLogLevel.Maximized);
// posSqlConnection.Open();
// setPumpOnStartingCommand.ExecuteNonQuery();
// }
// catch (Exception ex)
// {
// fccLogger.LogDebug("executing setPumpOnStartingCommand failed, exception detail: " + ex,
// DebugLogLevel.Normal);
// }
// }
//}
//else if (pumpStatusStr == "FUELLING")
//{
// /* indicate for fueling is ongoing */
//}
//else if (pumpStatusStr == "IDLE")
//{
// /* indicate for nozzle if replaced back */
// var sizeLevelNozzleIdsOnPump = SiteConfigUtility.Default.GetSiteLevelNozzleIdsByPumpId(p);
// if (!sizeLevelNozzleIdsOnPump.Any())
// {
// fccLogger.LogDebug("Could not found any site level nozzle ids for pump: " + p);
// return;
// }
// using (var posSqlConnection = new SqlConnection(ConfigurationManager.ConnectionStrings["PosDatabaseConnStr"].ConnectionString))
// {
// try
// {
// /* idle would not carry nozzle id, so here reset all nozzles on target pump.*/
// var setPumpOnIdleCommand
// = new SqlCommand(sizeLevelNozzleIdsOnPump.Select(siteLevelNozzleId =>
// {
// var totalizer = SiteConfigUtility.Default.GetTotalizer(siteLevelNozzleId);
// return
// string.Format(
// "Update jy_info set [status] = '{1}', qty=0, amount=0, fzqty={2}, fzamount={3}" +
// " where jihao = '{0}'", siteLevelNozzleId, 'F',
// totalizer.Item1, totalizer.Item2);
// })
// .Aggregate((acc, n) => acc + " " + n), posSqlConnection);
// fccLogger.LogDebug("setPumpOnIdleCommand: " + setPumpOnIdleCommand.CommandText);
// posSqlConnection.Open();
// setPumpOnIdleCommand.ExecuteNonQuery();
// }
// catch (Exception ex)
// {
// fccLogger.LogDebug("executing setPumpOnIdleCommand failed, exception detail: " + ex);
// }
// }
//}
}
public bool Start()
{
if (0 == Interlocked.CompareExchange(ref this.isStarted, 1, 0))
{
this.msgRouterClient.Start();
return true;
}
else
{
throw new InvalidOperationException("Already started.");
}
}
public bool IsStarted
{
get { return this.isStarted == 1; }
}
private void SendSubscriptions()
{
Thread.Sleep(100);
// pump events
this.msgRouterClient.SendMessage(MsgRouterMessageUtility.SubscribePumpStatusChange());
Thread.Sleep(100);
// pump delivery
this.msgRouterClient.SendMessage(MsgRouterMessageUtility.SubscribePumpDeliveryProgress());
Thread.Sleep(100);
// fuel price change events
this.msgRouterClient.SendMessage(MsgRouterMessageUtility.SubscribeNewPriceChangeApplied());
Thread.Sleep(100);
this.msgRouterClient.SendMessage(MsgRouterMessageUtility.SubscribeRefreshPriceChangeTbl());
Thread.Sleep(100);
this.msgRouterClient.SendMessage(MsgRouterMessageUtility.SubscribeSetNewPriceChange());
Thread.Sleep(100);
this.msgRouterClient.SendMessage(MsgRouterMessageUtility.SubscribeReceivePresetFromForecourt());
Thread.Sleep(100);
this.msgRouterClient.SendMessage(MsgRouterMessageUtility.SubscribeReceiveOutdoorTrxDeniedFromForecourt());
Thread.Sleep(100);
this.msgRouterClient.SendMessage(MsgRouterMessageUtility.SubscribeReceiveOutdoorTrxApprovedFromForecourt());
Thread.Sleep(100);
this.msgRouterClient.SendMessage(MsgRouterMessageUtility.SubscribeReceiveOutdoorTrxAirlockFromForecourt());
}
public void Dispose()
{
this.msgRouterClient.Dispose();
}
}
}