JiShe.CollectBus/src/JiShe.CollectBus.IoTDBProvider/Provider/SessionPoolAdapter.cs

using Apache.IoTDB.DataStructure;
using Apache.IoTDB;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using JiShe.CollectBus.IoTDBProvider.Interface;
using Microsoft.Extensions.Logging;

namespace JiShe.CollectBus.IoTDBProvider.Provider
{
    /// <summary>
    /// 树模型连接池
    /// </summary>
    public class SessionPoolAdapter : IIoTDBSessionPool
    {
        private readonly SessionPool _sessionPool;
        private readonly IoTDBOptions _options;

        public SessionPoolAdapter(IoTDBOptions options)
        {
            _options = options;
            _sessionPool = new SessionPool.Builder()
                .SetNodeUrl(options.ClusterList)
                .SetUsername(options.UserName)
                .SetPassword(options.Password)
                .SetFetchSize(options.FetchSize)
                .SetPoolSize(options.PoolSize)
                .Build();
        }

        /// <summary>
        /// 打开连接池
        /// </summary>
        /// <returns></returns>
        public async Task OpenAsync()
        {
            await _sessionPool.Open(false);
            if (_options.OpenDebugMode)
            {
                _sessionPool.OpenDebugMode(builder =>
                {
                    builder.AddConsole();
                });
            }
        }

        /// <summary>
        /// 批量插入对齐时间序列数据
        /// </summary>
        /// <param name="tablet"></param>
        /// <returns></returns>
        public async Task<int> InsertAsync(Tablet tablet)
        {
            var result =  await _sessionPool.InsertAlignedTabletAsync(tablet);
            return result;
        }

        /// <summary>
        /// 查询数据
        /// </summary>
        /// <param name="sql"></param>
        /// <returns></returns>
        public async Task<SessionDataSet> ExecuteQueryStatementAsync(string sql)
        {
            return await _sessionPool.ExecuteQueryStatementAsync(sql);
        }

        public void Dispose()
        {
            _sessionPool?.Close().ConfigureAwait(false).GetAwaiter().GetResult();
        }
    }
}