few more tests

Author: ideoma

src/net-questdb-client/LineTcpSender.cs

@@ -8,25 +8,22 @@ namespace QuestDB
     public class LineTcpSender : IDisposable
     {
         private static readonly long EpochTicks = new DateTime(1970, 1, 1).Ticks;
-        private readonly TcpClient _client;
-        private readonly bool _closeClient;
+        private readonly Socket _clientSocket;
         private readonly byte[] _sendBuffer;
         private int _position;
         private bool _hasMetric;
         private bool _quoted;
         private bool _noFields = true;
-
-        public LineTcpSender(TcpClient client, int bufferSize = 4096)
+        
+        public LineTcpSender(String address, int port, int bufferSize = 4096)
         {
-            _client = client;
+            _clientSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
+            _clientSocket.NoDelay = true;
+            _clientSocket.Blocking = true;
+            _clientSocket.Connect(address, port);
             _sendBuffer = new byte[bufferSize];
         }
 
-        public LineTcpSender(String address, int port, int bufferSize = 4096) : this(new TcpClient(address, port), bufferSize)
-        {
-            _closeClient = true;
-        }
-
         public LineTcpSender Metric(ReadOnlySpan<char> name)
         {
             if (_hasMetric)
@@ -141,9 +138,9 @@ private void PutUtf8(char c)
                 Flush();
             }
 
-            Span<byte> bytes = _sendBuffer;
+            Span<byte> bytes = _sendBuffer.AsSpan(_position);
             Span<char> chars = stackalloc char[1] {c};
-            _position += Encoding.UTF8.GetBytes(chars, bytes.Slice(_position, 4));
+            _position += Encoding.UTF8.GetBytes(chars, bytes);
         }
 
         private void PutSpecial(char c)
@@ -202,7 +199,8 @@ private LineTcpSender Put(char c)
 
         public void Flush()
         {
-            _position -= _client.Client.Send(_sendBuffer, 0, _position, SocketFlags.None);
+            int sent = _clientSocket.Send(_sendBuffer, 0, _position, SocketFlags.None);
+            _position -= sent;
         }
 
         public void Dispose()
@@ -220,10 +218,7 @@ public void Dispose()
             }
             finally
             {
-                if (_closeClient)
-                {
-                    _client.Dispose();
-                }
+                _clientSocket.Dispose();
             }
         }
 
@@ -237,7 +232,7 @@ public void AtNow()
         public void At(DateTime timestamp)
         {
             long epoch = timestamp.Ticks - EpochTicks;
-            Put(' ').Put(epoch).Put("00").AtNow();
+            Put(' ').Put(epoch).Put('0').Put('0').AtNow();
         }
     }
 }

src/tcp-client-test/LineTcpSenderTests.cs

@@ -1,4 +1,5 @@
 using System;
+using System.IO;
 using System.Net;
 using System.Net.Sockets;
 using System.Text;
@@ -40,8 +41,8 @@ public void SendLineExceedsBuffer()
             srv.AcceptAsync();
 
             using var ls = new LineTcpSender(IPAddress.Loopback.ToString(), _port, 25);
-            var lineCount = 50;
-            var expected = "metric\\ name,t\\ a\\ g=v\\ alu\\,\\ e number=10i,db\\ l=123.12,string=\" -=\\\"\",привед=\"медвед\" 1000000000\n";
+            var lineCount = 500;
+            var expected = "metric\\ name,t\\ a\\ g=v\\ alu\\,\\ e number=10i,db\\ l=123.12,string=\" -=\\\"\",при\\ вед=\"медвед\" 1000000000\n";
             var totalExpectedSb = new StringBuilder();
             for (int i = 0; i < lineCount; i++)
             {
@@ -50,7 +51,7 @@ public void SendLineExceedsBuffer()
                     .Field("number", 10)
                     .Field("db l", 123.12)
                     .Field("string", " -=\"")
-                    .Field("привед", "медвед")
+                    .Field("при вед", "медвед")
                     .At(new DateTime(1970, 01, 01, 0, 0, 1));
                 totalExpectedSb.Append(expected);
             }
@@ -78,6 +79,31 @@ public void SendNegativeLongAndDouble()
             var expected = "neg\\\\name number1=-9223372036854775807i,number2=9223372036854775807i,number3=-1.7976931348623157E+308,number4=1.7976931348623157E+308\n";
             WaitAssert(srv, expected);
         }
+        
+        [Test]
+        public void SendMillionToFile()
+        {
+            using var srv = CreateTcpListener(_port);
+            srv.AcceptAsync();
+
+            var  nowMillisecond = DateTime.Now.Millisecond;
+            var metric = "metric_name" + nowMillisecond;
+
+            using var ls = new LineTcpSender(IPAddress.Loopback.ToString(), _port, 2048);
+            for(int i = 0; i < 1E6; i++)
+            {
+                ls.Metric(metric)
+                    .Tag("nopoint", "tag" + i%100 )
+                    .Field("counter", i * 1111.1)
+                    .Field("int", i)
+                    .Field("привед", "мед вед")
+                    .At(new DateTime(2021, 1, 1, (i/360/1000) % 60, (i/60/1000) % 60, (i / 1000) % 60, i % 1000));
+            }
+            ls.Flush();
+            
+            File.WriteAllText($"out-{nowMillisecond}.txt", srv.GetTextReceived()); 
+        }
+
 
         private static void WaitAssert(DummyIlpServer srv, string expected)
         {
@@ -177,8 +203,8 @@ private class DummyIlpServer : IDisposable
             private readonly TcpListener _server;
             private readonly byte[] _buffer = new byte[2048];
             private readonly CancellationTokenSource _cancellationTokenSource = new();
-            private readonly StringBuilder _request = new();
-            private volatile int _totalReceived = 0; 
+            private volatile int _totalReceived;
+            private readonly MemoryStream _received = new();
 
             public DummyIlpServer(int port)
             {
@@ -211,9 +237,8 @@ private async Task SaveData(Socket connection)
                     int received = await connection.ReceiveAsync(_buffer, SocketFlags.None, _cancellationTokenSource.Token);
                     if (received > 0)
                     {
-                        var value = Encoding.UTF8.GetString(_buffer, 0, received);
-                        _request.Append(value);
-                        _totalReceived += value.Length;
+                        _received.Write(_buffer, 0, received);
+                        _totalReceived += received;
                     }
                 } 
             }
@@ -228,7 +253,7 @@ public void Dispose()
 
             public string GetTextReceived()
             {
-                return _request.ToString();
+                return Encoding.UTF8.GetString(_received.GetBuffer(), 0, (int)_received.Length);
             }
         }
     }