using System;
using System.Collections.Generic;
using System.Configuration;
using System.Data;
using System.Data.Odbc;
using System.Data.SqlClient;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading;
using System.Timers;
using Wayne.Lib;
using Wayne.Lib.Log;
namespace SinochemInternetPlusApp
{
internal class SiteConfigScannerCommunicator : ICommunicator, IDisposable
{
static NLog.Logger debugLogger = NLog.LogManager.LoadConfiguration("nlog.config").GetLogger("PumpHandler");
/// <summary>
/// 0 for not started, 1 for started already.
/// </summary>
private int isStarted = 0;
//private readonly DebugLogger debugLogger =
// new DebugLogger(new IdentifiableEntity(0, "FC2PosProxyMain", "", null));
/// <summary>
/// PumpIdAsSideA:PumpIdAsSideB:JiHao
/// </summary>
private readonly IEnumerable<Tuple<int, int, int>> pumpSidePairs = null;
/// <summary>
/// pumpId:RealHoseLogicalId:HuiTianSiteLevelHoseId
/// </summary>
public static IEnumerable<Tuple<int, int, int>> nozzleRemappingPairs = null;
private System.Timers.Timer scannerTimer;
/// <summary>
/// Start a site configuration scanner and try sync the FC db with POS's.
/// call start() for this communicator will force a sync right now.
/// </summary>
/// <param name="scanningInterval"></param>
public SiteConfigScannerCommunicator(int scanningInterval)
{
var pumpSideMappingRawString = GenericSinochemEpsApp.AppSettings["PumpSideMapping"].Trim().Replace(" ", "");
if (pumpSideMappingRawString.Substring(pumpSideMappingRawString.Length - 1) == ";")
pumpSideMappingRawString = pumpSideMappingRawString.Substring(0, pumpSideMappingRawString.Length - 1);
this.pumpSidePairs = pumpSideMappingRawString.Split(';').Select(p =>
new Tuple<int, int, int>(int.Parse(p.Split(':')[0]), int.Parse(p.Split(':')[1]),
int.Parse(p.Split(':')[2])));
if (GenericSinochemEpsApp.AppSettings["forceMappingFusionHoseToHuiTianHose"] != null)
{
var nozzleRemappingRawString = GenericSinochemEpsApp.AppSettings["forceMappingFusionHoseToHuiTianHose"].Trim().Replace(" ", "");
if (nozzleRemappingRawString.Substring(nozzleRemappingRawString.Length - 1) == ";")
nozzleRemappingRawString = nozzleRemappingRawString.Substring(0, nozzleRemappingRawString.Length - 1);
debugLogger.Info("will parse forceMappingFusionHoseToHuiTianHose, raw string in config is: " + nozzleRemappingRawString, DebugLogLevel.Maximized);
nozzleRemappingPairs = nozzleRemappingRawString.Split(';').Select(p =>
new Tuple<int, int, int>(int.Parse(p.Split(':')[0]), int.Parse(p.Split(':')[1]),
int.Parse(p.Split(':')[2])));
}
this.scannerTimer = new System.Timers.Timer(scanningInterval);
this.scannerTimer.Elapsed += scannerTimer_Elapsed;
}
void scannerTimer_Elapsed(object sender, ElapsedEventArgs e)
{
this.PerformCompareAndUpdate();
}
/// <summary>
/// read config from FC database, and check if need update to POS database.
/// </summary>
/// <returns>true for updated, false for not updated.</returns>
private bool PerformCompareAndUpdate()
{
try
{
var configInFcDb = this.LoadOrderedConfigFromFcDb();
var configInPosDb = this.LoadOrderedConfigFromPosDb();
if (configInFcDb == null || !configInFcDb.Any())
{
debugLogger.Info("config In FC Database is empty, skip this round");
return false;
}
SiteConfigUtility.Default.UpdateLatestSiteConfig(configInFcDb);
// for now, always overwrite the whole table in POS database without further check for rows or columns level.
using (var posConn =
new SqlConnection(GenericSinochemEpsApp.AppSettings["PosDatabaseConnStr"]))
{
/* configInFcDb structure like below */
//" JiHao: " + oneRow["jiHao"] + ", sideId: " + oneRow["sideId"] +
//", nozzleLogicalId: " + oneRow["HoseLogicalId"] +
//", siteLevelNozzleId: " + oneRow["siteLevelNozzleId"]
string bulkInsertCmd = configInFcDb.Select(
r => string.Format("Select {0}, '{1}', {2}, {3}",
r["jiHao"],
r["sideId"],
r["HoseLogicalId"],
r["siteLevelNozzleId"])).Aggregate((acc, n) => acc + " Union all " + n);
var truncateAndInsertCmd =
new SqlCommand(
"TRUNCATE table jyjpz; INSERT jyjpz (jihao, abtype, qianghao, luojiqh) " + bulkInsertCmd,
posConn);
debugLogger.Info("truncateAndInsertSiteConfigCmd: " + truncateAndInsertCmd.CommandText);
posConn.Open();
truncateAndInsertCmd.ExecuteNonQuery();
return true;
}
}
catch (Exception ex)
{
debugLogger.Error("PerformCompareAndUpdate exceptioned, detail: " + ex);
return false;
}
}
public bool IsStarted
{
get { return this.isStarted == 1; }
}
public bool Start()
{
if (0 == Interlocked.CompareExchange(ref this.isStarted, 1, 0))
{
// the initial load must be succeed!
if (!this.PerformCompareAndUpdate())
{
return false;
}
this.scannerTimer.Start();
return true;
}
else
{
throw new InvalidOperationException("Already started.");
}
}
private IEnumerable<DataRow> LoadOrderedConfigFromFcDb()
{
try
{
using (var fcOdbcConnection = new SqlConnection(GenericSinochemEpsApp.AppSettings["FCDatabaseConnStr"]))
{
using (var oda = new SqlDataAdapter(
"select id as PumpId, device_id as HoseName, param_value as HoseLogicalId " +
" from config_values where library = 'pump' and device_id like 'hose%' and parameter ='logical_id' " +
" order by id::int, device_id",
fcOdbcConnection))
{
var siteNozzlesDataTableInFcDb = new DataTable();
oda.Fill(siteNozzlesDataTableInFcDb);
//MemoryStream ms = new MemoryStream();
//siteNozzlesDataTableInFcDb.WriteXml(ms);
//StreamReader sr = new StreamReader(ms);
//var rawData = sr.ReadToEnd();
//debugLogger.Add("LoadConfigFromFcDb original data: \r\n" + rawData, DebugLogLevel.Normal);
/* the returned data from postgresql db is like:
PumpId HoseName HoseLogicalId
"1" "hose3" "3"
"1" "hose4" "4"
"1" "hose2" "2"
"1" "hose1" "1"
"10" "hose2" "2"
"10" "hose1" "1"
"10" "hose3" "3"
"10" "hose4" "4"
*/
// PetroChina defined a id of: 'jiHao' which stands for a physical pump(2 sides).
siteNozzlesDataTableInFcDb.Columns.Add(new DataColumn("jiHao"));
// PetroChina need another id of site overall id for a nozzle, start from 1.
// we cacualte this based on jiHao and HostLogicalId
siteNozzlesDataTableInFcDb.Columns.Add(new DataColumn("siteLevelNozzleId"));
// as doc defined, side A is 0, side B is 1.
siteNozzlesDataTableInFcDb.Columns.Add(new DataColumn("sideId"));
var siteNozzlesDataRowsInFcDb = siteNozzlesDataTableInFcDb.Rows.Cast<DataRow>();
foreach (var row in siteNozzlesDataRowsInFcDb)
{
var pumpId = int.Parse(row["pumpid"].ToString());
var possibleSideA = this.pumpSidePairs.FirstOrDefault(p => p.Item1 == pumpId);
Tuple<int, int, int> possibleSideB = null;
if (possibleSideA != null)
{
row["sideId"] = "0";
row["jiHao"] = possibleSideA.Item3;
}
else if ((possibleSideB = this.pumpSidePairs.FirstOrDefault(p => p.Item2 == pumpId)) != null)
{
row["sideId"] = "1";
row["jiHao"] = possibleSideB.Item3;
}
else
{
throw new ArgumentException("Pump with pumpId: " + pumpId +
" is neither side A nor side B, pls check Side definition config file.");
}
}
var sortedSiteNozzlesDataRowsInFcDb =
siteNozzlesDataRowsInFcDb.OrderBy(r => int.Parse(r["jiHao"].ToString()))
.ThenBy(r => r["sideId"])
.ThenBy(r => r["HoseLogicalId"]);
debugLogger.Debug("sortedSiteNozzlesDataRowsInFcDb: ");
// finally we get the ordered sequence, now give them sitelevel nozzle id.
for (var i = 0; i < sortedSiteNozzlesDataRowsInFcDb.Count(); i++)
{
var oneRow = sortedSiteNozzlesDataRowsInFcDb.ElementAt(i);
oneRow["siteLevelNozzleId"] = i + 1;
debugLogger.Debug(" JiHao: " + oneRow["jiHao"] + ", sideId: " + oneRow["sideId"] +
", nozzleLogicalId: " + oneRow["HoseLogicalId"] +
", siteLevelNozzleId: " + oneRow["siteLevelNozzleId"]
+ ", pumpId: " + oneRow["pumpid"]
//+ ", hoseName: " + oneRow["hosename"]
);
if (nozzleRemappingPairs != null)
{
var pumpId = int.Parse(oneRow["pumpid"].ToString());
var fusionHoseLogicalId = int.Parse(oneRow["HoseLogicalId"].ToString());
var matched = nozzleRemappingPairs.FirstOrDefault(m => m.Item1 == pumpId && m.Item2 == fusionHoseLogicalId);
if (matched != null)
{
debugLogger.Debug("Found nozzleRemappingPairs for pumpId: " + pumpId + " with HoseLogicalId: " + fusionHoseLogicalId + ", " +
"will remapping the bound siteLevelNozzleId from: " + oneRow["siteLevelNozzleId"] + " to: " + matched.Item3);
oneRow["siteLevelNozzleId"] = matched.Item3;
}
}
}
return sortedSiteNozzlesDataRowsInFcDb;
}
}
}
catch (Exception ex)
{
debugLogger.Error("LoadConfigFromFcDb exceptioned, detail: " + ex);
throw;
}
}
private IEnumerable<DataRow> LoadOrderedConfigFromPosDb()
{
try
{
using (var posConn = new SqlConnection(GenericSinochemEpsApp.AppSettings["PosDatabaseConnStr"]))
{
using (var sda = new SqlDataAdapter("select * from jyjpz order by jihao, abtype, qianghao", posConn))
{
var siteNozzlesDataTableInPosDb = new DataTable();
sda.Fill(siteNozzlesDataTableInPosDb);
debugLogger.Debug("siteNozzlesDataTableInPosDb: ");
for (var i = 0; i < siteNozzlesDataTableInPosDb.Rows.Count; i++)
{
var oneRow = siteNozzlesDataTableInPosDb.Rows[i];
debugLogger.Debug(" JiHao: " + oneRow["jihao"] + ", sideId(abtype): " + oneRow["abtype"]
+ ", nozzleLogicalId(qianghao): " + oneRow["qianghao"] +
", siteLevelNozzleId(luojiqh): " + oneRow["luojiqh"]);
}
return siteNozzlesDataTableInPosDb.Rows.Cast<DataRow>();
}
}
}
catch (Exception ex)
{
debugLogger.Error("LoadConfigFromPosDb exceptioned, detail: " + ex);
throw;
}
}
public void Dispose()
{
this.scannerTimer.Stop();
}
}
}