release300.2

Author: dolphindb-jffu

include/ConstantImp.h

@@ -936,7 +936,7 @@ class AbstractFastVector: public Vector{
 		if(buf == (Y*)data_+start)
 			return true;
 		else if(getType() == sourceType)
-			memcpy(data_+start,buf,sizeof(Y)*len);
+			memcpy(reinterpret_cast<void *>(data_ + start), reinterpret_cast<const void *>(buf), sizeof(Y) * len);
 		else
 			for(int i=0;i<len;++i)
 				data_[start+i]=(buf[i] == sourceNullVal)? nullVal_: static_cast<T>(buf[i]);
@@ -947,7 +947,7 @@ class AbstractFastVector: public Vector{
 	inline bool appendData(Y* buf, int len, DATA_TYPE sourceType, Y sourceNullVal){
 		checkCapacity(len);
 		if(getType() == sourceType)
-			memcpy(data_+size_, buf, sizeof(Y) * len);
+			memcpy(reinterpret_cast<void *>(data_ + size_), reinterpret_cast<const void *>(buf), sizeof(Y) * len);
 		else
 			for(int i=0;i<len;++i)
 				data_[size_+i]=(buf[i] == sourceNullVal)? nullVal_ : static_cast<T>(buf[i]);

include/DBConnectionPoolImpl.h

@@ -7,15 +7,16 @@ namespace dolphindb {
 class DBConnectionPoolImpl{
 public:
     struct Task{
-        Task(const std::string& sc = "", int id = 0, int pr = 4, int pa = 2, bool clearM = false)
-                : script(sc), identity(id), priority(pr), parallelism(pa), clearMemory(clearM){}
-        Task(const std::string& function, const std::vector<ConstantSP>& args, int id = 0, int pr = 4, int pa = 2, bool clearM = false) 
-            : script(function), arguments(args), identity(id), priority(pr), parallelism(pa), clearMemory(clearM){ isFunc = true; }
+        Task(const std::string& sc = "", int id = 0, int pr = 4, int pa = 2, int fetchSize = 0, bool clearM = false)
+                : script(sc), identity(id), priority(pr), parallelism(pa), fetchSize(fetchSize), clearMemory(clearM) {}
+        Task(const std::string& function, const std::vector<ConstantSP>& args, int id = 0, int pr = 4, int pa = 2, int fetchSize = 0, bool clearM = false)
+            : script(function), arguments(args), identity(id), priority(pr), parallelism(pa), fetchSize(fetchSize), clearMemory(clearM){ isFunc = true; }
         std::string script;
         std::vector<ConstantSP> arguments;
         int identity;
         int priority;
         int parallelism;
+        int fetchSize;
         bool clearMemory;
         bool isFunc = false;
     };
@@ -33,12 +34,12 @@ class DBConnectionPoolImpl{
         }
     }
     void run(const std::string& script, int identity, int priority=4, int parallelism=64, int fetchSize=0, bool clearMemory = false){
-        queue_->push(Task(script, identity, priority, parallelism, clearMemory));
+        queue_->push(Task(script, identity, priority, parallelism, fetchSize, clearMemory));
         taskStatus_.setResult(identity, TaskStatusMgmt::Result());
     }
 
     void run(const std::string& functionName, const std::vector<ConstantSP>& args, int identity, int priority=4, int parallelism=64, int fetchSize=0, bool clearMemory = false){
-        queue_->push(Task(functionName, args, identity, priority, parallelism, clearMemory));
+        queue_->push(Task(functionName, args, identity, priority, parallelism, fetchSize, clearMemory));
         taskStatus_.setResult(identity, TaskStatusMgmt::Result());
     }
 

include/SharedMem.h

@@ -1,5 +1,5 @@
 #ifndef __SharedMem_H
-#define __IpcMem_H
+#define __SharedMem_H
 
 #ifdef LINUX
 

include/Streaming.h

@@ -10,6 +10,15 @@ namespace dolphindb {
 class DBConnection;
 class StreamingClientImpl;
 
+struct SubscribeQueue {
+	MessageQueueSP queue_;
+	std::shared_ptr<std::atomic<bool>> stopped_{nullptr};
+	SubscribeQueue()
+		:queue_(nullptr), stopped_(nullptr) {}
+	SubscribeQueue(MessageQueueSP queue, std::shared_ptr<std::atomic<bool>> stopped)
+		:queue_(queue), stopped_(stopped) {}
+};
+
 class EXPORT_DECL StreamingClient {
 public:
 	//listeningPort > 0 : listen mode, wait for server connection
@@ -20,7 +29,7 @@ class EXPORT_DECL StreamingClient {
 	void exit();
 
 protected:
-    MessageQueueSP subscribeInternal(std::string host, int port, std::string tableName, std::string actionName = DEFAULT_ACTION_NAME,
+    SubscribeQueue subscribeInternal(std::string host, int port, std::string tableName, std::string actionName = DEFAULT_ACTION_NAME,
                                      int64_t offset = -1, bool resubscribe = true, const VectorSP &filter = nullptr,
                                      bool msgAsTable = false, bool allowExists = false, int batchSize  = 1,
 									 std::string userName="", std::string password="",

src/AsynWorker.cpp

@@ -22,10 +22,10 @@ namespace dolphindb {
         while(true) {
             try {
                 if(task.isFunc){
-                    result = conn_->run(task.script, task.arguments, task.priority, task.parallelism, 0, task.clearMemory);
+                    result = conn_->run(task.script, task.arguments, task.priority, task.parallelism, task.fetchSize, task.clearMemory);
                 }
                 else{
-                    result = conn_->run(task.script, task.priority, task.parallelism, 0, task.clearMemory);
+                    result = conn_->run(task.script, task.priority, task.parallelism, task.fetchSize, task.clearMemory);
                 }
                 break;
             }

src/CMakeLists.txt

@@ -0,0 +1,42 @@
+if (UNIX)
+    target_compile_definitions(${CMAKE_PROJECT_NAME} PUBLIC LINUX)
+endif()
+
+target_include_directories(${CMAKE_PROJECT_NAME}
+    PUBLIC ${PROJECT_SOURCE_DIR}/include
+    PUBLIC ${OPENSSL_INCLUDE_DIR}
+)
+
+target_sources(${CMAKE_PROJECT_NAME} PUBLIC
+    AsynWorker.cpp
+    BatchTableWriter.cpp
+    Compress.cpp
+    Concurrent.cpp
+    ConstantImp.cpp
+    ConstantMarshall.cpp
+    DBConnectionPoolImpl.cpp
+    DFSChunkMeta.cpp
+    DictionaryImp.cpp
+    DolphinDB.cpp
+    DolphinDBImp.cpp
+    DomainImp.cpp
+    EventHandler.cpp
+    Exceptions.cpp
+    Format.cpp
+    Guid.cpp
+    int128.cpp
+    LZ4.cpp
+    Matrix.cpp
+    MultithreadedTableWriter.cpp
+    ScalarImp.cpp
+    SetImp.cpp
+    SharedMem.cpp
+    Streaming.cpp
+    StreamingUtil.cpp
+    SymbolBase.cpp
+    SysIO.cpp
+    TableImp.cpp
+    TaskStatusMgmt.cpp
+    Util.cpp
+    Vector.cpp
+)

src/Concurrent.cpp

@@ -460,7 +460,7 @@ void Thread::setAffinity(int id) {
 	CPU_ZERO(&mask);
 	CPU_SET(id, &mask);
 	if (sched_setaffinity(0, sizeof(mask), &mask) == -1) {
-		throw RuntimeException("BindCore failed, error code " + errno);
+		throw RuntimeException("BindCore failed, error code " + std::to_string(errno));
 	}
 #endif
 }

src/MultithreadedTableWriter.cpp

@@ -46,7 +46,7 @@ MultithreadedTableWriter::MultithreadedTableWriter(const std::string& hostName,
     if (pCompressMethods != nullptr && pCompressMethods->size() > 0) {
         for (auto one : *pCompressMethods) {
             if (one != COMPRESS_DELTA && one != COMPRESS_LZ4) {
-                throw RuntimeException("Unsupported compression method " + one);
+                throw RuntimeException("Unsupported compression method " + std::to_string(one));
             }
         }
         saveCompressMethods_ = *pCompressMethods;