update

Author: windancer2010

.gitignore

@@ -34,3 +34,8 @@ dolphindb_csharpapi/obj/
 *.cache
 *.cache
 dolphindb_csharpapi_test/obj/Release/CoreCompileInputs.cache
+dolphindb_csharpapi/bin/Release/dolphindb_csharpapi.instr.pdb
+dolphindb_csharpapi/bin/Release/dolphindb_csharpapi.pdb
+dolphindb_csharpapi/dolphindb_csharpapi.csproj.user
+dolphindb_csharpapi/dolphindb_csharpapi.csproj.user
+*.user

dolphindb_csharpapi/bin/Release/dolphindb_csharpapi.dll

@@ -67,8 +67,12 @@
     <Compile Include="data\BasicFloatMatrix.cs" />
     <Compile Include="data\BasicFloatVector.cs" />
     <Compile Include="data\BasicInt.cs" />
+    <Compile Include="data\BasicInt128.cs" />
+    <Compile Include="data\BasicInt128Vector.cs" />
     <Compile Include="data\BasicIntMatrix.cs" />
     <Compile Include="data\BasicIntVector.cs" />
+    <Compile Include="data\BasicIPAddr.cs" />
+    <Compile Include="data\BasicIPAddrVector.cs" />
     <Compile Include="data\BasicLong.cs" />
     <Compile Include="data\BasicLongMatrix.cs" />
     <Compile Include="data\BasicLongVector.cs" />
@@ -102,6 +106,8 @@
     <Compile Include="data\BasicTimestampMatrix.cs" />
     <Compile Include="data\BasicTimestampVector.cs" />
     <Compile Include="data\BasicTimeVector.cs" />
+    <Compile Include="data\BasicUuid.cs" />
+    <Compile Include="data\BasicUuidVector.cs" />
     <Compile Include="data\IDictionary.cs" />
     <Compile Include="data\IEntity.cs" />
     <Compile Include="data\IEntityFactory.cs" />
@@ -123,6 +129,7 @@
     <Compile Include="io\ExtendedDataOutput.cs" />
     <Compile Include="io\LittleEndianDataInputStream.cs" />
     <Compile Include="io\LittleEndianDataOutputStream.cs" />
+    <Compile Include="io\Long2.cs" />
     <Compile Include="io\ProgressListener.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
     <Compile Include="RSAUtils.cs" />

dolphindb_csharpapi/dolphindb_csharpapi.csproj

@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="utf-8"?>
 <Project ToolsVersion="15.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
   <PropertyGroup>
-    <ProjectView>ProjectFiles</ProjectView>
+    <ProjectView>ShowAllFiles</ProjectView>
   </PropertyGroup>
 </Project>

dolphindb_csharpapi/dolphindb_csharpapi.csproj.user

@@ -13,6 +13,7 @@ public abstract class AbstractClient : MessageDispatcher
         protected static readonly int DEFAULT_PORT = 8849;
         protected static readonly string DEFAULT_HOST = "localhost";
         protected static readonly string DEFAULT_ACTION_NAME = "csharpStreamingApi";
+        protected string listeningHost;
         protected int listeningPort;
         protected QueueManager queueManager = new QueueManager();
         protected Dictionary<string, List<IMessage>> messageCache = new Dictionary<string, List<IMessage>>();
@@ -85,7 +86,9 @@ private void activeCloseConnection(Site site)
                     conn.connect(site.host, site.port);
                     try
                     {
-                        string localIP = conn.LocalAddress;
+                        string localIP = listeningHost;
+                        if (localIP == null || localIP.Equals(String.Empty))
+                            localIP = conn.LocalAddress;
                         List<IEntity> @params = new List<IEntity>
                         {
                             new BasicString(localIP),
@@ -121,6 +124,14 @@ public AbstractClient(int subscribePort)
             pThread.Start();
         }
 
+        public AbstractClient(string subscribeHost, int subscribePort)
+        {
+            listeningHost = subscribeHost;
+            listeningPort = subscribePort;
+            Deamon deamon = new Deamon(subscribePort, this);
+            pThread = new Thread(new ThreadStart(deamon.run));
+            pThread.Start();
+        }
         private void addMessageToCache(IMessage msg)
         {
             string topic = msg.getTopic();
@@ -201,7 +212,9 @@ protected BlockingCollection<List<IMessage>> subscribeInternal(string host, int
             dbConn.connect(host, port);
             try
             {
-                string localIP = dbConn.LocalAddress;
+                string localIP = listeningHost;
+                if (localIP == null || localIP.Equals(String.Empty))
+                    localIP = dbConn.LocalAddress;
 
                 if (!hostEndian.ContainsKey(host))
                     hostEndian.Add(host, dbConn.RemoteLittleEndian);

dolphindb_csharpapi/streaming/AbstractClient.cs

@@ -14,6 +14,8 @@ public class PollingClient : AbstractClient
 
         public PollingClient(int subscribePort) : base(subscribePort) { }
 
+        public PollingClient(string subscribeHost, int subscribePort) : base(subscribeHost, subscribePort) { }
+
         override protected void doReconnect(Site site)
         {
             while (true)

dolphindb_csharpapi/streaming/PollingClient.cs

@@ -43,6 +43,49 @@ public TaskPool(int threadsMaxCount)
 
         public ThreadPooledClient() : this(DEFAULT_PORT) { }
 
+        public ThreadPooledClient(string subscribeHost, int subscribePort) : base(subscribeHost,subscribePort) {
+            Queue<IMessage> backlog = new Queue<IMessage>();
+
+            bool fillBacklog()
+            {
+                bool filled = false;
+                lock (queueHandlers)
+                {
+                    foreach (QueueHandlerBinder binder in queueHandlers.Values)
+                    {
+                        if (binder.Item1.TryTake(out List<IMessage> messages))
+                        {
+                            messages.ForEach(backlog.Enqueue);
+                            filled = true;
+                        }
+                    }
+                }
+                return filled;
+            }
+
+            void run()
+            {
+                while (true)
+                {
+                    while (backlog.Count > 0)
+                    {
+                        IMessage msg = backlog.Dequeue();
+                        QueueHandlerBinder binder;
+                        lock (queueHandlers)
+                        {
+                            binder = queueHandlers[msg.getTopic()];
+                        }
+                        HandlerRunner handlerRunner = new HandlerRunner(binder.Item2, msg);
+                        ThreadPool.QueueUserWorkItem(new WaitCallback(handlerRunner.run));
+                    }
+                    fillBacklog();
+                }
+            }
+
+            thread = new Thread(new ThreadStart(run));
+            thread.Start();
+        }
+
         public ThreadPooledClient(int subscribePort) : base(subscribePort)
         {
             Queue<IMessage> backlog = new Queue<IMessage>();

dolphindb_csharpapi/streaming/ThreadPooledClient.cs

@@ -15,6 +15,9 @@ public ThreadedClient() : this(DEFAULT_PORT) { }
 
         public ThreadedClient(int subscribePort) : base(subscribePort) { }
 
+        public ThreadedClient(string subscribeHost, int subscribePort) : base(subscribeHost, subscribePort) { }
+
+
         class HandlerLooper
         {
             BlockingCollection<List<IMessage>> queue;