ZKLT/Hadoop/ZKLT.Hadoop/HadoopService.cs

using Microsoft.AspNetCore.Builder;
using Microsoft.Extensions.DependencyInjection;
using MySqlX.XDevAPI.Relational;
using Newtonsoft.Json.Linq;
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;

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>();
            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();
        }
    }
}