You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
495 lines
17 KiB
C#
495 lines
17 KiB
C#
using Microsoft.AspNetCore.Builder;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using MySqlX.XDevAPI.Relational;
|
|
using Newtonsoft.Json.Linq;
|
|
using Quartz.Impl;
|
|
using Quartz;
|
|
using Quartz.Spi;
|
|
using System;
|
|
using System.Collections.Generic;
|
|
using System.Linq;
|
|
using System.Text;
|
|
using System.Threading.Tasks;
|
|
using System.Transactions;
|
|
using ZKLT.Hadoop.Interface;
|
|
using ZKLT.Hadoop.Model;
|
|
using ZKLT.Quartz;
|
|
using ZKLT.Quartz.Interface;
|
|
|
|
namespace ZKLT.Hadoop
|
|
{
|
|
/// <summary>
|
|
/// 云计算扩展
|
|
/// </summary>
|
|
public static class HadoopServiceExtend
|
|
{
|
|
/// <summary>
|
|
/// 注入云计算服务
|
|
/// </summary>
|
|
/// <param name="services">服务集合</param>
|
|
/// <returns></returns>
|
|
public static IServiceCollection AddHadoop(this IServiceCollection services)
|
|
{
|
|
services.AddSingleton<IHadoopService, HadoopService>();
|
|
services.AddSingleton<ITableService, TableService>();
|
|
services.AddSingleton<ITaskService, TaskService>();
|
|
services.AddSingleton<IQuartzService, QuartzService>();
|
|
|
|
return services;
|
|
}
|
|
|
|
/// <summary>
|
|
/// 应用云计算
|
|
/// </summary>
|
|
/// <param name="app">应用</param>
|
|
/// <param name="config">源配置</param>
|
|
/// <returns></returns>
|
|
public static IApplicationBuilder UseHadoop(this IApplicationBuilder app, Action<HDP_Source> config)
|
|
{
|
|
var _TableService = app.ApplicationServices.GetRequiredService<ITableService>();
|
|
_TableService.Init(config);
|
|
return app;
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// 云计算服务
|
|
/// </summary>
|
|
public class HadoopService : IHadoopService
|
|
{
|
|
public HadoopService(ITableService tableService)
|
|
{
|
|
_TableService = tableService;
|
|
}
|
|
|
|
private ITableService _TableService;
|
|
|
|
/// <summary>
|
|
/// 创建源
|
|
/// </summary>
|
|
/// <param name="source">源</param>
|
|
/// <returns>是否成功</returns>
|
|
public bool InsertSource(HDP_Source source)
|
|
{
|
|
//校验
|
|
if (string.IsNullOrEmpty(source.Id))
|
|
{
|
|
throw new ArgumentNullException("编号无效");
|
|
}
|
|
if (string.IsNullOrEmpty(source.Host))
|
|
{
|
|
throw new ArgumentNullException("主机无效");
|
|
}
|
|
if (source.Port == null)
|
|
{
|
|
throw new ArgumentNullException("端口无效");
|
|
}
|
|
if (string.IsNullOrEmpty(source.Key))
|
|
{
|
|
throw new ArgumentNullException("数据源键无效");
|
|
}
|
|
if (string.IsNullOrEmpty(source.Account))
|
|
{
|
|
throw new ArgumentNullException("用户名无效");
|
|
}
|
|
if (string.IsNullOrEmpty(source.PassWord))
|
|
{
|
|
throw new ArgumentNullException("密码无效");
|
|
}
|
|
if (_TableService.GetSource(source.Id) != null)
|
|
{
|
|
throw new ArgumentException("编号已存在");
|
|
}
|
|
|
|
return _TableService.Insert("", "HDP_Source", HDP_Table.Class2JObject(source));
|
|
}
|
|
|
|
/// <summary>
|
|
/// 更新源
|
|
/// </summary>
|
|
/// <param name="source">源</param>
|
|
/// <returns>是否成功</returns>
|
|
public bool UpdateSource(HDP_Source source)
|
|
{
|
|
//校验
|
|
if (string.IsNullOrEmpty(source.Id))
|
|
{
|
|
throw new ArgumentNullException("编号无效");
|
|
}
|
|
if (string.IsNullOrEmpty(source.Host))
|
|
{
|
|
throw new ArgumentNullException("主机无效");
|
|
}
|
|
if (source.Port == null)
|
|
{
|
|
throw new ArgumentNullException("端口无效");
|
|
}
|
|
if (string.IsNullOrEmpty(source.Key))
|
|
{
|
|
throw new ArgumentNullException("数据源键无效");
|
|
}
|
|
if (string.IsNullOrEmpty(source.Account))
|
|
{
|
|
throw new ArgumentNullException("用户名无效");
|
|
}
|
|
if (string.IsNullOrEmpty(source.PassWord))
|
|
{
|
|
throw new ArgumentNullException("密码无效");
|
|
}
|
|
|
|
return _TableService.Update(source.Id, "HDP_Source", new JObject {
|
|
{"Id","=" }
|
|
}, HDP_Table.Class2JObject(source));
|
|
}
|
|
|
|
/// <summary>
|
|
/// 删除源
|
|
/// </summary>
|
|
/// <param name="sourceid">源</param>
|
|
/// <returns>是否成功</returns>
|
|
public bool DeleteSource(string sourceid)
|
|
{
|
|
return _TableService.Delete(sourceid, "HDP_Source", new JObject{
|
|
{"Id","=" }
|
|
}, new JObject{
|
|
{"Id",sourceid }
|
|
});
|
|
}
|
|
|
|
/// <summary>
|
|
/// 查询源
|
|
/// </summary>
|
|
/// <param name="command">命令</param>
|
|
/// <returns>结果</returns>
|
|
public HDP_Source[] QuerySource(HDP_Command command)
|
|
{
|
|
return _TableService.Query<HDP_Source>(command.SourceId!, command.TableId!, command.Where!, command.Data!, command.Order!, command.Col);
|
|
}
|
|
|
|
|
|
/// <summary>
|
|
/// 创建表
|
|
/// </summary>
|
|
/// <param name="table">表</param>
|
|
/// <returns>是否成功</returns>
|
|
public bool InsertTable(HDP_Table table)
|
|
{
|
|
try
|
|
{
|
|
if (_TableService.GetTable(table.Id) != null)
|
|
{
|
|
throw new ArgumentNullException("表编号已存在");
|
|
}
|
|
}
|
|
catch (ArgumentNullException e)
|
|
{
|
|
if (!e.Message.Contains("数据表不存在"))
|
|
{
|
|
throw;
|
|
}
|
|
}
|
|
|
|
if (_TableService.InitStruct("", table))
|
|
{
|
|
using (TransactionScope _scope = new TransactionScope())
|
|
{
|
|
if (!_TableService.Insert("", "HDP_Table", HDP_Table.Class2JObject(table)))
|
|
{
|
|
return false;
|
|
}
|
|
for (var i = 0; i < table.Columns!.Length; i++)
|
|
{
|
|
var _column = table.Columns![i];
|
|
_column.TableId = table.Id;
|
|
if (!_TableService.Insert("", "HDP_Column", HDP_Table.Class2JObject(_column)))
|
|
{
|
|
return false;
|
|
}
|
|
}
|
|
_scope.Complete();
|
|
return true;
|
|
}
|
|
}
|
|
else
|
|
{
|
|
return false;
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// 更新表
|
|
/// </summary>
|
|
/// <param name="table">表</param>
|
|
/// <returns>是否成功</returns>
|
|
public bool UpdateTable(HDP_Table table)
|
|
{
|
|
if (_TableService.InitStruct("", table))
|
|
{
|
|
using (TransactionScope _scope = new TransactionScope())
|
|
{
|
|
if (!_TableService.Update("", "HDP_Table", new JObject{
|
|
{ "Id","="}
|
|
}, HDP_Table.Class2JObject(table)))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
for (var i = 0; i < table.Columns!.Length; i++)
|
|
{
|
|
var _column = table.Columns![i];
|
|
_column.TableId = table.Id;
|
|
if (_TableService.QuerySingle<HDP_Column>("", "HDP_Column", new JObject
|
|
{
|
|
{"Id","=" }
|
|
}, HDP_Table.Class2JObject(_column), null) == null)
|
|
{
|
|
if (!_TableService.Insert("", "HDP_Column", HDP_Table.Class2JObject(_column)))
|
|
{
|
|
return false;
|
|
}
|
|
}
|
|
else
|
|
{
|
|
if (!_TableService.Update("", "HDP_Column", new JObject {
|
|
{"Id","=" }
|
|
}, HDP_Table.Class2JObject(_column)))
|
|
{
|
|
return false;
|
|
}
|
|
}
|
|
}
|
|
|
|
_scope.Complete();
|
|
return true;
|
|
}
|
|
}
|
|
else
|
|
{
|
|
return false;
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// 删除表
|
|
/// </summary>
|
|
/// <param name="tableId">表编号</param>
|
|
/// <returns>是否成功</returns>
|
|
public bool DeleteTable(string tableId)
|
|
{
|
|
//校验
|
|
if (string.IsNullOrEmpty(tableId))
|
|
{
|
|
throw new ArgumentNullException("编号无效");
|
|
}
|
|
|
|
using (TransactionScope _scope = new TransactionScope())
|
|
{
|
|
if (!_TableService.Delete("", "HDP_Table", new JObject{
|
|
{"Id","=" }
|
|
}, new JObject {
|
|
{"Id",tableId }
|
|
}))
|
|
{
|
|
return false;
|
|
}
|
|
if (!_TableService.Delete("", "HDP_Column", new JObject{
|
|
{ "TableId","="}
|
|
}, new JObject {
|
|
{"TableId",tableId }
|
|
}))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
_scope.Complete();
|
|
return true;
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// 查询表
|
|
/// </summary>
|
|
/// <param name="command">命令</param>
|
|
/// <returns>结果</returns>
|
|
public HDP_Table[] QueryTable(HDP_Command command)
|
|
{
|
|
return _TableService.Query<HDP_Table>("", "HDP_Table", command.Where!, command.Data!, command.Order!, command.Col!);
|
|
}
|
|
|
|
/// <summary>
|
|
/// 插入数据
|
|
/// </summary>
|
|
/// <param name="command">命令</param>
|
|
/// <returns>是否成功</returns>
|
|
public bool Insert(HDP_Command command)
|
|
{
|
|
return _TableService.Insert(command.SourceId!, command.TableId!, command.Data!);
|
|
}
|
|
|
|
/// <summary>
|
|
/// 更新数据
|
|
/// </summary>
|
|
/// <param name="command"></param>
|
|
/// <returns>是否成功</returns>
|
|
public bool Update(HDP_Command command)
|
|
{
|
|
return _TableService.Update(command.SourceId!, command.TableId!, command.Where!, command.Data!);
|
|
}
|
|
|
|
/// <summary>
|
|
/// 删除数据
|
|
/// </summary>
|
|
/// <param name="command">命令</param>
|
|
/// <returns>是否成功</returns>
|
|
public bool Delete(HDP_Command command)
|
|
{
|
|
return _TableService.Delete(command.SourceId!, command.TableId!, command.Where!, command.Data!);
|
|
}
|
|
|
|
/// <summary>
|
|
/// 查询单条
|
|
/// </summary>
|
|
/// <typeparam name="T">返回类型</typeparam>
|
|
/// <param name="command">命令</param>
|
|
/// <returns>结果</returns>
|
|
public T? QuerySingle<T>(HDP_Command command)
|
|
{
|
|
return _TableService.QuerySingle<T>(command.SourceId!, command.TableId!, command.Where!, command.Data!, command.Col);
|
|
}
|
|
|
|
/// <summary>
|
|
/// 查询列表
|
|
/// </summary>
|
|
/// <typeparam name="T">返回类型</typeparam>
|
|
/// <param name="command">命令</param>
|
|
/// <returns>结果</returns>
|
|
public T[] Query<T>(HDP_Command command)
|
|
{
|
|
return _TableService.Query<T>(command.SourceId!, command.TableId!, command.Where!, command.Data!, command.Order!, command.Col);
|
|
}
|
|
|
|
/// <summary>
|
|
/// 分页查询
|
|
/// </summary>
|
|
/// <typeparam name="T">返回类型</typeparam>
|
|
/// <param name="command">命令</param>
|
|
/// <returns>结果</returns>
|
|
public HDP_Page<T> Page<T>(HDP_Command command)
|
|
{
|
|
if (command.PageIndex == null || command.PageIndex <= 0)
|
|
{
|
|
throw new ArgumentNullException("分页下标无效");
|
|
}
|
|
if (command.PageSize == null || command.PageSize <= 0)
|
|
{
|
|
throw new ArgumentNullException("分页大小无效");
|
|
}
|
|
return _TableService.QueryPage<T>(command.SourceId!, command.TableId!, (int)command.PageIndex, (int)command.PageSize, command.Where!, command.Data!, command.Order!, command.Col!);
|
|
}
|
|
|
|
/// <summary>
|
|
/// 批量执行任务
|
|
/// </summary>
|
|
/// <param name="command">指令</param>
|
|
/// <returns></returns>
|
|
public object?[] PatchCommand(HDP_Command[] command)
|
|
{
|
|
object?[] _result = new object?[command.Length];
|
|
using (TransactionScope _scope = new TransactionScope())
|
|
{
|
|
var _isComplete = true;
|
|
for (var i = 0; i < command.Length; i++)
|
|
{
|
|
try
|
|
{
|
|
if (command[i].Type == HDP_CommandType.INSERT)
|
|
{
|
|
var _temp = Insert(command[i]);
|
|
if (!_temp)
|
|
{
|
|
_isComplete = false;
|
|
}
|
|
_result[i] = _temp;
|
|
}
|
|
else if (command[i].Type == HDP_CommandType.UPDATE)
|
|
{
|
|
var _temp = Update(command[i]);
|
|
if (!_temp)
|
|
{
|
|
_isComplete = false;
|
|
}
|
|
_result[i] = _temp;
|
|
}
|
|
else if (command[i].Type == HDP_CommandType.DELETE)
|
|
{
|
|
var _temp = Delete(command[i]);
|
|
if (!_temp)
|
|
{
|
|
_isComplete = false;
|
|
}
|
|
_result[i] = _temp;
|
|
}
|
|
else if (command[i].Type == HDP_CommandType.QUERYSINGLE || command[i].Type == HDP_CommandType.QUERY || command[i].Type == HDP_CommandType.PAGE)
|
|
{
|
|
continue;
|
|
}
|
|
else
|
|
{
|
|
_result[i] = "不支持该命令";
|
|
_isComplete = false;
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_result[i] = ex.Message;
|
|
_isComplete = false;
|
|
}
|
|
}
|
|
if (_isComplete)
|
|
{
|
|
_scope.Complete();
|
|
}
|
|
}
|
|
for (var i = 0; i < command.Length; i++)
|
|
{
|
|
try
|
|
{
|
|
if (command[i].Type == HDP_CommandType.INSERT || command[i].Type == HDP_CommandType.UPDATE || command[i].Type == HDP_CommandType.DELETE)
|
|
{
|
|
continue;
|
|
}
|
|
else if (command[i].Type == HDP_CommandType.QUERYSINGLE)
|
|
{
|
|
var _temp = QuerySingle<dynamic>(command[i]);
|
|
_result[i] = _temp;
|
|
}
|
|
else if (command[i].Type == HDP_CommandType.QUERY)
|
|
{
|
|
var _temp = Query<dynamic>(command[i]);
|
|
_result[i] = _temp;
|
|
}
|
|
else if (command[i].Type == HDP_CommandType.PAGE)
|
|
{
|
|
var _temp = Page<dynamic>(command[i]);
|
|
_result[i] = _temp;
|
|
}
|
|
else
|
|
{
|
|
_result[i] = "不支持该命令";
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_result[i] = ex.Message;
|
|
}
|
|
}
|
|
return _result.ToArray();
|
|
}
|
|
|
|
public T? StoredProcedure<T>(HDP_Command command)
|
|
{
|
|
return _TableService.StoredProcedure<T>(command.SourceId!, command.StoredProcedureName!, command.Data!);
|
|
}
|
|
}
|
|
}
|