插件开发深度解析

编写插件时，DolphinDB 中的大部分数据对象都可以用 Constant 类型来表示（标量、向量、矩阵、表，等等），使用时调用 ConstantSP 进行操作，ConstantSP 是一个经过封装的智能指针，会在变量的引用计数为 0 时自动释放内存，不需要用户手动释放。从它派生的其它常用变量类型有：VectorSP（向量）、TableSP（表）等。

在编写插件时往往需要对 Vector 和内存表中的数据进行读写。DolphinDB 提供了许多读写的接口函数，如果使用不当将影响读写效率，因此下面将介绍如何使用 DolphinDB 提供的这些接口函数实现对 Vector 和内存表的高效读写。

编写插件时，如果需要创建线程，不建议使用 pthread_create 等函数直接创建线程，这样创建的线程没有初始化 DolphinDB 中的一些随机数，在执行线程函数时可能会出错。建议使用 new 语句创建在头文件 Concurrent.h 中声明的 Thread 对象，它需要传入一个 RunnableSP 对象，所以需要用户实现一个 Runnable 对象的派生类，并完成纯虚函数 void run() 的实现，用 void run() 来执行线程函数，然后将 Runnable 派生类对象的 SmartPointer 传给 Thread。本例中 DemoRun 为 Runnable 的派生类，void run() 函数读取一个整型数组的数据并打印，createThread 函数创建了一个线程通过 start() 函数来执行 DemoRun 的 run() 函数，并 join() 回收该线程：

class DemoRun : public Runnable {
public:
    DemoRun(ConstantSP data) : data_(data) {}
    ~DemoRun() {}
    void run() override {
        size_t size = data_->size();
        for (size_t i = 0; i < size; ++i) {
            std::cout << data_->getInt(i) << std::endl;
        }
    }
private:
    ConstantSP data_;
};

ConstantSP createThread(Heap *heap, vector<ConstantSP> &arguments) {
    if (!(arguments[0]->isVector() && arguments[0]->getType() == DT_INT)) {
        throw IllegalArgumentException("createThread", "argument must be an integral vector");
    }
    SmartPointer<DemoRun> demoRun = new DemoRun(arguments[0]);
    ThreadSP thread = new Thread(demoRun);
    if (!thread->isStarted()) {
        thread->start();
    }
    thread->join();
    return new Void();
}

由于 join() 函数会阻塞等待子线程的退出，如果子线程需要长时间后台运行，则上述方法不合理。子线程运行时需要保证 ThreadSP 对象仍然存在，因此可以新建一个类，将 ThreadSP 作为其成员，通过 new 语句创建这个类，并用其创建子线程并执行线程函数，然后通过返回 Util::createResource 创建的对象来管理这个类的资源释放，因此需要实现一个回调函数 demoOnClose，当需要回收资源时会自动调用这个回调函数释放资源，用户也可以通过其他函数来手动释放资源：

class DemoRun2 : public Runnable {
public:
    DemoRun2(ConstantSP data) : data_(data) {}
    ~DemoRun2() {}
    void run() override {
        size_t size = data_->size();
        for (size_t i = 0; i < size; ++i) {
            std::cout << data_->getInt(i) << std::endl;
            Util::sleep(2000);
        }
    }
private:
    ConstantSP data_; 
};

class Demo {
public:
    Demo() {}
    ~Demo() {}
    void createAndRun(ConstantSP data) {
        SmartPointer<DemoRun2> demoRun = new DemoRun2(data);
        thread_ = new Thread(demoRun);
        if (!thread_->isStarted()) {
            thread_->start();
        }
    }
private:
    ThreadSP thread_;
};

static void demoOnClose(Heap* heap, vector<ConstantSP>& args) {
    Demo* pDemo = (Demo*)(args[0]->getLong());
    if(pDemo != nullptr) {
        delete pDemo;
        args[0]->setLong(0);
    }
}

ConstantSP createThread2(Heap *heap, vector<ConstantSP> &arguments) {
    if (!(arguments[0]->isVector() && arguments[0]->getType() == DT_INT)) {
        throw IllegalArgumentException("createThread", "argument must be an integral vector");
    }
    Demo * pDemo = new Demo();
    pDemo->createAndRun(arguments[0]);
    FunctionDefSP onClose(Util::createSystemProcedure("demo onClose()", demoOnClose, 1, 1));
    return Util::createResource(reinterpret_cast<long long>(pDemo), "Demo", onClose, heap->currentSession());
}

在编写插件获取 MQTT、OPC_UA、Kafka 等中间件的实时数据时，往往需要订阅这些数据，然后将数据写入 Dolphindb 的表中。由于订阅数据函数需要长时间连续运行，所以通常需要用插件中创建后台线程一节介绍的方法创建新线程来处理订阅消息。当用户订阅流数据表时，需要确保用户具有向保存流数据的表写入数据的权限，所以我们还需要设置用户权限，否则订阅流数据表时会出错。下面介绍如何在插件创建新线程并设置用户权限。

设置用户权限的前提是在创建新线程时创建一个新的会话。会话是一个容器，它具有唯一的 ID 并存储许多已经定义的对象，例如局部变量、共享变量等。创建新会话的方式有多种，如启动命令行窗口、XDB 连接、GUI 连接或 Web URL 连接。会话中的所有变量对于其他会话是不可见的，除非使用语句 share 在会话之间显式共享变量，目前 DolphinDB 仅支持表共享。

创建新线程时，线程函数如果需要用到会话的一些资源（比如调用函数或者访问权限），则需要创建一个新的会话，否则当前会话关闭后（比如在 GUI 里调用，然后关闭 GUI），这些资源自动释放后，如果线程函数仍在执行，会导致程序崩溃。因此在创建新线程时需要创建一个新的会话，即头文件 CoreConcept.h 中声明的 Session 对象，然后通过调用其成员函数 setUser 来设置权限。而每个 Session 都有一个 Output 类负责输出，而插件中 Output 通常不需要处理输出，所以可以实现一个如下所示的 DummyOutput 作为 Output 的派生类。

本例中 appendTable 为 Runnable 的派生类，在构造函数中用 heap->currentSession()->copy() 创建了一个新的会话 session_，并通过 setUser 设置权限，setOutput 设置输出，run 函数每隔 1s 往内存表添加数据，stopRun 函数用于停止线程；Client 负责创建线程并执行线程函数；subscribe 为用户接口函数，第一个参数为 table，第二个参数为 table 或回调函数，用户调用 subscribe 会每隔 1s 将第一个 table 添加到第二个 table 中（实际订阅第三方数据时，将第一个参数改成订阅的数据），cancelThread 用于停止线程函数：

    class DummyOutput: public Output{
    public:
        virtual bool timeElapsed(long long nanoSeconds){return true;}
        virtual bool write(const ConstantSP& obj){return true;}
        virtual bool message(const string& msg){return true;}
        virtual void enableIntermediateMessage(bool enabled) {}
        virtual IO_ERR done(){return OK;}
        virtual IO_ERR done(const string& errMsg){return OK;}
        virtual bool start(){return true;}
        virtual bool start(const string& message){return true;}
        virtual IO_ERR writeReady(){return OK;}
        virtual ~DummyOutput(){}
        virtual OUTPUT_TYPE getOutputType() const {return STDOUT;}
        virtual void close() {}
        virtual void setWindow(INDEX index,INDEX size){};
        virtual IO_ERR flush() {return OK;}
    };

    class Client;
    class appendTable : public Runnable {
    public:
        appendTable(Heap *heap, TableSP table, ConstantSP handle, Client* client) 
        : heap_(heap), table_(table), handle_(handle), client_(client) {
            session_ = heap->currentSession()->copy();//创建一个新的会话
            session_->setUser(heap->currentSession()->getUser());//设置权限
            session_->setOutput(new DummyOutput);//设置输出
        }
        ~appendTable() {}
        void run() override {
            while(true) {
                Util::-sleep(1000);
                if(handle_->isTable()) {
                    TableSP result = handle_;
                    std::vector<ConstantSP> dataToAppend = {result, table_};
                    session_->getFunctionDef("append!")->call(session_->getHeap().get(), dataToAppend);
                }
                else{
                    std::vector<ConstantSP> dataToAppend = {table_};
                    ((FunctionDefSP)handle_)->call(session_->getHeap().get(), dataToAppend);
                }
            }
        }
    private:
        Heap* heap_;
        SessionSP session_;
        Client * client_;
        ConstantSP handle_;
        TableSP table_;
    };

    class Client {
    public:
        Client() {}
        ~Client() {}
        void createAndRun(Heap *heap, ConstantSP table, ConstantSP handle) { 
            SmartPointer<appendTable> append = new appendTable(heap, table, handle, this);
            thread_ = new Thread(append);
            if (!thread_->isStarted()) {
                thread_->start();
            }
        }
        void cancelThread(){
            thread_->cancel();
        }
        
    private:
        ThreadSP thread_;
    };

    static void clientOnClose(Heap* heap, vector<ConstantSP>& args) {
        Client* pClient = (Client*)(args[0]->getLong());
        if(pClient != nullptr) {
            delete pClient;
            args[0]->setLong(0);
        }
    }

    ConstantSP subscribe(Heap *heap, vector<ConstantSP> &arguments) {
        if (!arguments[0]->isTable()) {
            throw IllegalArgumentException("subscribe", "First argument must be a table!");
        }
        if (!arguments[1]->isTable() && (arguments[1]->getType() != DT_FUNCTIONDEF)){
            throw IllegalArgumentException("subscribe", "Second argument must be a table or function!");
        }
        Client * pClient = new Client();
        pClient->createAndRun(heap, arguments[0], arguments[1]);
        FunctionDefSP onClose(Util::createSystemProcedure("client onClose()", clientOnClose, 1, 1));
        return Util::createResource(reinterpret_cast<long long>(pClient), "client", onClose, heap->currentSession());
    }

    ConstantSP cancelThread(Heap *heap, vector<ConstantSP> &arguments) {
        if (arguments[0]->getType() != DT_RESOURCE) {
            throw IllegalArgumentException("stopRun", "Argument must be a resource!");
        }
        Client* pClient = (Client*)(arguments[0]->getLong());
        pClient->cancelThread();
        return new Void();
    }