如何使用 Proxygen 和 Folly 发送 HTTP 分块响应以模拟视频流？

Posted 2023-03-15

【中文标题】如何使用 Proxygen 和 Folly 发送 HTTP 分块响应以模拟视频流？

【英文标题】：How to send HTTP chunked response to emulate a video stream using Proxygen and Folly?

【发布时间】：2015-06-15 14:02:43

【问题描述】：

我正在编写一个基于 Facebook Proxygen 的 HTTP 视频流服务器。没有计划的寻求。使用proxygen::ResponseBuilder，我能够将 webm 编码的视频块作为 HTTP 响应发送，即分块传输编码正在工作。我的问题是，Proxygen 甚至在发送响应标头之前就等待proxygen::ResponseBuilder::sendWithEOM()。我希望它在每次调用proxygen::ResponseBuilder::send() 后尽快实际发送数据。

我尝试使用 evb->runInLoop() 和 evb->runInEventBaseThread() 从 EventBaseThread 执行的 lambda 运行 ResponseBuilder 调用

using namespace folly;
using namespace proxygen;

std::thread t([&]()
    EventBase* evb = EventBaseManager::get()->getExistingEventBase();    
    // send headers ...
    while ( chunks avail. ) 
        //...
        evb->runInLoop([&]()
            ResponseBuilder(downstream_)
                     .body(std::move(chunk))
                     .send();
        );
        //... 
    
    // sendWithEOM ...
);
t.detach();

这段代码是从我的RequestHandler 的onRequest() 方法调用的。我尝试调用ResponseBuilder::send() 而不将其包装到evb->runInLoop() 中，但是带有Folly v0.42.0 的Proxygen v0.25.0 禁止使用断言从另一个线程调用ResponseBuilder::send()。我从这里删除了这个断言：https://github.com/facebook/folly/blob/v0.42.0/folly/io/async/EventBase.cpp#L491。

现在模拟流正在运行，但如果有并行请求，它就会崩溃。我想它不应该像这样使用，这就是断言的用途。但也许有人知道如何为我的用例正确使用 Proxygen 基础设施？

【参考方案1】：

这也有同样的问题。我得到它与这样的东西一起工作。

folly::EventBase* eventBase = folly::EventBaseManager::get()->getExistingEventBase();
thread t([&, eventBase]() 
    while( chunks exist ) 
        auto chunk = getChunk();    
        eventBase->runInEventBaseThread([&, chunk=chunk]() mutable 
            ResponseBuilder(downstream_).body(move(chunk)).send();
        );
    
);
// sendWithEOM ...
t.detach();

【讨论】：

由于我需要更多的声望给+1，谢谢你的回答，为我解决了。

【参考方案2】：

using namespace folly;
using namespace proxygen;

//Get Event Base in worker thread and pass it to new thread. If you call this inside new thread then you won't get worker thread's event base.
EventBase* evb = EventBaseManager::get()->getExistingEventBase();   
std::thread t([&, evb]()

// send headers ...
while ( chunks avail. ) 
    //...
    evb->runInLoop([&]()
        ResponseBuilder(downstream_)
                 .body(std::move(chunk))
                 .send();
    );
    //... 

// sendWithEOM ...
);
t.detach();

因此无需在 EventBase 源中注释断言。