c++ - 将数据从 OpenCV C++ 传递到 NodeJS/JS | Electron

Question

我正在尝试使用以下堆栈为视频处理应用程序做一个 POC，并且对将处理后的媒体流从 C++ 应用程序传递到 Electron 前端 GUI 感到震惊。

  Electron  
     |
   Nodejs
     |
C++ Application

C++ 应用程序将读取 IP/网络摄像头(仅使用 OpenCV 获取数据)并处理输入流(不使用 OpenCV)。我正在尝试找出一种方法，以良好的 fps 将该流从 C++ 发送到 Electron GUI(NodeJS/JS)。现在，我使用 node-gyp 编译了我的 C++ 应用程序并将其安装为 Node 包。

此外，我不想过多地更改我的 C++ 应用程序(比如将 OpenCV 作为 Node 包)，因为稍后我将单独使用该 C++ 应用程序与另一个应用程序集成。

Answer 1

挑战:

我们希望在单独的工作线程中执行繁重的代码，同时在执行期间将结果(流数据 block )发送回主线程。

NAN(Node.js 的 native 抽象)已经提供了一种使用 (AsyncProgressWorker) 执行此操作的方法。

但是，我们无法知道在执行期间是否实际调用了 HandleProgressCallback 以发送回我们的结果。当我们的运行时间太快并且因此永远不会执行回调时，就会发生这种情况。

建议的解决方案:

我们只是将流输出收集到堆栈 (StackCollect) 中。我们尝试立即清除此堆栈并将流结果发送回主线程(如果可能)- (StackDrain)。如果我们没有时间立即清除堆栈，我们会在执行运行 (HandleOKCallback) 结束时清空(剩下的)。

实现示例:

demo.cpp(我们的 C++ Node/Electron 插件):

#include <nan.h>
#include <node.h>
#include <v8.h>

#include <iostream>
#include <string>
#include <vector>

#include <mutex>

#include <chrono>
#include <thread>

class vSync_File : public Nan::AsyncProgressWorker {
public:
    ~vSync_File();
    vSync_File(Nan::Callback * result, Nan::Callback * chunk);

    void Execute(const Nan::AsyncProgressWorker::ExecutionProgress& chunk);

    void HandleOKCallback();

    void HandleProgressCallback(const char *tout, size_t tout_size);

    //needed for stream data collection
    void StackCollect(std::string & str_chunk, const Nan::AsyncProgressWorker::ExecutionProgress& tchunk);
    //drain stack
    void StackDrain();
private:
    Nan::Callback * chunk;
    //stores stream data - use other data types for different output
    std::vector<std::string> stack;
    //mutex
    std::mutex m;
};

vSync_File::vSync_File(Nan::Callback * result, Nan::Callback * chunk)
: Nan::AsyncProgressWorker(result), chunk(chunk) {}

vSync_File::~vSync_File() {
    delete chunk;
}

void vSync_File::StackCollect(std::string & str_chunk, const Nan::AsyncProgressWorker::ExecutionProgress& tchunk) { 
    std::lock_guard<std::mutex> guardme(m);

    stack.push_back(str_chunk);

    //attempt drain
    std::string dummy = "NA";
    tchunk.Send(dummy.c_str(), dummy.length());
}

//Dump out stream data
void vSync_File::StackDrain() {
  std::lock_guard<std::mutex> guardme(m);

  for (uint i = 0; i < stack.size(); i++) {
    std::string th_chunk = stack[i];
    v8::Local<v8::String> chk = Nan::New<v8::String>(th_chunk).ToLocalChecked();
    v8::Local<v8::Value> argv[] = { chk };

    chunk->Call(1, argv, this->async_resource);
  }
  stack.clear();
}

//Our main job in a nice worker thread
void vSync_File::Execute(const Nan::AsyncProgressWorker::ExecutionProgress& tchunk) {
    //simulate some stream output
    for (unsigned int i = 0; i < 20; i++) {
        std::string out_chunk;
        out_chunk = "Simulated stream data " + std::to_string(i);   

        std::this_thread::sleep_for(std::chrono::milliseconds(300)); //so our HandleProgressCallback is invoked, otherwise we are too fast in our example here

        this->StackCollect(out_chunk, tchunk);
    }   
}

//Back at the main thread - if we have time stream back the output 
void vSync_File::HandleProgressCallback(const char *tout, size_t tout_size) {
  Nan::HandleScope scope;

  this->StackDrain();
}

//Back at the main thread - we are done
void vSync_File::HandleOKCallback () {

  this->StackDrain(); //drain leftovers from stream stack

  v8::Local<v8::String> result_mess = Nan::New<v8::String>("done reading").ToLocalChecked();
  v8::Local<v8::Value> argv[] = { result_mess };
  callback->Call(1, argv, this->async_resource);
}


NAN_METHOD(get_stream_data) {
  Nan::Callback *result = new Nan::Callback(info[0].As<v8::Function>());
  Nan::Callback *chunk = new Nan::Callback(info[1].As<v8::Function>());

  AsyncQueueWorker(new vSync_File(result, chunk));
}


NAN_MODULE_INIT(Init) {
   //we want stream data
   Nan::Set(target, Nan::New<v8::String>("get_stream_data").ToLocalChecked(),
   Nan::GetFunction(Nan::New<v8::FunctionTemplate>(get_stream_data)).ToLocalChecked());
}

NODE_MODULE(stream_c_electron, Init)

index.js( Electron 实现示例):

const stream_c_electron = require('./build/linux_x64/stream_c_electron.node');

stream_c_electron.get_stream_data(function(res) {
    //we are done
    console.log(res);
}, function(chk) {
    console.log("a line streamed");
    console.log(chk);
});

package.json:

{
  "name": "stream_c_electron",
  "version": "1.0.0",
  "description": "stream from c++ node addon demo",
  "main": "index.js",
  "scripts": {
    "start": "electron .",
    "build_this": "HOME=~/.electron-gyp node-gyp rebuild --target=2.0.8 --arch=x64 --dist-url=https://atom.io/download/electron",
    "test": "echo \"Error: no test specified\" && exit 1"
  },
  "author": "11AND2",
  "license": "MIT",
  "dependencies": {
    "nan": "2.11.0"
  },
  "devDependencies": {
    "electron": "2.0.8"
  }
}

绑定(bind).gyp:

{
  "targets": [
    {
      "target_name": "stream_c_electron",
      "sources": [  "c_src/demo.cpp"  ],
      "conditions": [
        [
        'OS=="linux"',
        {
            "cflags": ["-Wall", "-std=c++11"],
            'product_dir' : 'linux_x64',
            "include_dirs": [
                "<!(node -e \"require('nan')\")"
            ]
        }
        ]
      ]
    }
   ]
}