Application.cpp

Go to the documentation of this file.

 
#include "rogue/Directives.h"
 
#include "rogue/protocols/packetizer/Application.h"
 
#include <memory>
 
#include "rogue/GeneralError.h"
#include "rogue/GilRelease.h"
#include "rogue/Logging.h"
#include "rogue/interfaces/stream/Buffer.h"
#include "rogue/interfaces/stream/Frame.h"
#include "rogue/protocols/packetizer/Controller.h"
 
namespace rpp = rogue::protocols::packetizer;
namespace ris = rogue::interfaces::stream;
 
#ifndef NO_PYTHON
    #include <boost/python.hpp>
namespace bp = boost::python;
#endif
 
rpp::ApplicationPtr rpp::Application::create(uint8_t id) {
    rpp::ApplicationPtr r = std::make_shared<rpp::Application>(id);
    return (r);
}
 
void rpp::Application::setup_python() {
#ifndef NO_PYTHON
 
    bp::class_<rpp::Application, rpp::ApplicationPtr, bp::bases<ris::Master, ris::Slave>, boost::noncopyable>(
        "Application",
        bp::init<uint8_t>());
 
    bp::implicitly_convertible<rpp::ApplicationPtr, ris::MasterPtr>();
    bp::implicitly_convertible<rpp::ApplicationPtr, ris::SlavePtr>();
#endif
}
 
rpp::Application::Application(uint8_t id) {
    id_ = id;
    queue_.setMax(8);
}
 
rpp::Application::~Application() {
    // No-op if setController() never ran.
    if (thread_) {
        threadEn_ = false;
        rogue::GilRelease noGil;
        queue_.stop();
        thread_->join();
    }
}
 
void rpp::Application::setController(rpp::ControllerPtr cntl) {
    // One-shot: re-assigning a joinable std::thread triggers std::terminate.
    if (thread_) return;
 
    // threadEn_ must precede thread ctor; rollback both on spawn failure.
    rpp::ControllerPtr prevCntl = cntl_;
    cntl_ = cntl;
    threadEn_ = true;
    try {
        thread_ = std::make_unique<std::thread>(&rpp::Application::runThread, this);
    } catch (...) {
        threadEn_ = false;
        cntl_ = prevCntl;
        throw;
    }
 
    // Set a thread name
#ifndef __MACH__
    pthread_setname_np(thread_->native_handle(), "PackApp");
#endif
}
 
ris::FramePtr rpp::Application::acceptReq(uint32_t size, bool zeroCopyEn) {
    return (cntl_->reqFrame(size));
}
 
void rpp::Application::acceptFrame(ris::FramePtr frame) {
    cntl_->applicationRx(frame, id_);
}
 
void rpp::Application::pushFrame(ris::FramePtr frame) {
    queue_.push(frame);
}
 
void rpp::Application::runThread() {
    ris::FramePtr frame;
    Logging log("packetizer.Application");
    log.logThreadId();
 
    while (threadEn_) {
        if ((frame = queue_.pop()) != NULL) sendFrame(frame);
    }
}