Merge pull request #78 from maikebing/main

优化Rowrecord的构造函数，为SessionPool与SessionDataset添加IDisposable pattern

Author: lausannel

src/Apache.IoTDB/DataStructure/RowRecord.cs

@@ -9,8 +9,12 @@ public class RowRecord
     {
         public long Timestamps { get; }
         public List<object> Values { get; }
-        public List<string> Measurements { get; }
-
+        public List<string> Measurements { get; }
+
+        public RowRecord(DateTime timestamp, List<object> values, List<string> measurements)
+            :this(new DateTimeOffset(timestamp.ToUniversalTime()).ToUnixTimeMilliseconds(), values,measurements)
+        {
+        }
         public RowRecord(long timestamps, List<object> values, List<string> measurements)
         {
             Timestamps = timestamps;

src/Apache.IoTDB/DataStructure/SessionDataSet.cs

@@ -6,7 +6,7 @@
 
 namespace Apache.IoTDB.DataStructure
 {
-    public class SessionDataSet
+    public class SessionDataSet: System.IDisposable
     {
         private readonly long _queryId;
         private readonly string _sql;
@@ -23,8 +23,8 @@ public class SessionDataSet
         private int _rowIndex;
         private bool _hasCatchedResult;
         private RowRecord _cachedRowRecord;
-        private readonly bool _isClosed = false;
-
+        private   bool _isClosed = false;
+        private bool disposedValue;
 
         private string TimestampStr => "Time";
         private int StartIndex => 2;
@@ -295,7 +295,7 @@ public async Task Close()
 
                 try
                 {
-                    await myClient.ServiceClient.closeOperationAsync(req);
+                  var status=  await myClient.ServiceClient.closeOperationAsync(req);
                 }
                 catch (TException e)
                 {
@@ -306,6 +306,34 @@ public async Task Close()
                     _clientQueue.Add(myClient);
                 }
             }
-        }
+        }
+
+        protected virtual void Dispose(bool disposing)
+        {
+            if (!disposedValue)
+            {
+                if (disposing)
+                {
+                    try
+                    {
+                        this.Close().Wait();
+                    }
+                    catch 
+                    {
+                    }
+                }
+                _queryDataset=null; 
+                _timeBuffer = null;
+                _valueBufferLst = null;
+                _bitmapBufferLst = null;
+                disposedValue = true;
+            }
+        }
+
+        public void Dispose()
+        {
+            Dispose(disposing: true);
+            GC.SuppressFinalize(this);
+        }
     }
 }

src/Apache.IoTDB/SessionPool.cs

@@ -14,7 +14,8 @@
 
 namespace Apache.IoTDB
 {
-    public class SessionPool
+
+    public class SessionPool:IDisposable
     {
         private static int SuccessCode => 200;
         private static readonly TSProtocolVersion ProtocolVersion = TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V3;
@@ -89,7 +90,9 @@ public SessionPool(
             string password = "root",
             int fetchSize = 1000,
             string zoneId = "UTC+08:00",
-            int poolSize = 8)
+            int poolSize = 8,
+            bool enableRpcCompression = true
+            )
         {
             _host = host;
             _port = port;
@@ -99,8 +102,12 @@ public SessionPool(
             _fetchSize = fetchSize;
             _debugMode = false;
             _poolSize = poolSize;
-        }
-        ILoggerFactory factory;
+            _enableRpcCompression = enableRpcCompression;
+        }
+
+        ILoggerFactory factory;
+        private bool disposedValue;
+
         public void OpenDebugMode(Action<ILoggingBuilder> configure)
         {
             _debugMode = true;
@@ -111,16 +118,22 @@ public void OpenDebugMode(Action<ILoggingBuilder> configure)
         public void CloseDebugMode()
         {
             _debugMode = false;
-        }
-
+        }
+
         public async Task Open(bool enableRpcCompression, CancellationToken cancellationToken = default)
+        {
+            _enableRpcCompression = enableRpcCompression;
+            await Open(cancellationToken);
+        }
+
+        public async Task Open(CancellationToken cancellationToken = default)
         {
-            _clients = new ConcurrentClientQueue();
-            _enableRpcCompression = enableRpcCompression;
-
+            _clients = new ConcurrentClientQueue();
+
+
             for (var index = 0; index < _poolSize; index++)
             {
-                _clients.Add(await CreateAndOpen(enableRpcCompression, cancellationToken));
+                _clients.Add(await CreateAndOpen(_enableRpcCompression, cancellationToken));
             }
         }
 
@@ -2081,15 +2094,28 @@ public async Task<List<string>> ShowPathsTemplateUsingOnAsync(string templateNam
             {
                 _clients.Add(client);
             }
-        }
-
-
-
-
-
-
-
-
-
+        }
+
+        protected virtual void Dispose(bool disposing)
+        {
+            if (!disposedValue)
+            {
+                if (disposing)
+                {
+#if NET461_OR_GREATER || NETSTANDARD2_0
+#else
+                    _clients.ClientQueue.Clear();
+#endif
+                }
+                _clients = null;
+                disposedValue = true;
+            }
+        }
+
+        public void Dispose()
+        {
+            Dispose(disposing: true);
+            GC.SuppressFinalize(this);
+        }
     }
 }