This commit is contained in:
2015-06-08 16:25:24 +02:00
parent f925e43d1f
commit d6c4b72278
27 changed files with 1075 additions and 235 deletions
@@ -112,6 +112,13 @@ celix_status_t bundleActivator_stop(void * userData, bundle_context_pt context)
celix_status_t status = CELIX_ILLEGAL_STATE;
if (userData != nullptr) {
UserData* myUserData = (UserData*)userData;
if (myUserData->webResourceReg != nullptr) {
serviceRegistration_unregister(myUserData->webResourceReg);
myUserData->webResourceReg = nullptr;
}
if (myUserData->webResourceImpl) {
myUserData->webResourceImpl.reset();
}
if (myUserData->serviceReg != nullptr) {
serviceRegistration_unregister(myUserData->serviceReg);
myUserData->serviceReg = nullptr;
@@ -12,10 +12,6 @@
#include <Poco/Net/WebSocket.h>
#include "Poco/Net/NetException.h"
#include <Poco/Net/StreamSocket.h>
#include <sys/socket.h>
#include <sys/select.h>
#include <sys/time.h>
#include <stropts.h>
@@ -288,103 +284,6 @@ void DeviceRepoHandler::__handleCommand(Poco::Net::HTTPServerRequest &req, Poco:
void DeviceRepoHandler::__handleEventLongPolling(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp, DeviceEventQueue* eventQueue) {
Variant msg;
bool breakLoop = false;
boost::asio::deadline_timer connCheckTimer(parent->getIOService());
Net::HTTPServerRequestImpl* reqImpl = static_cast<Net::HTTPServerRequestImpl*>(&req);
std::mutex _m;
std::function<void(const boost::system::error_code& error)> connCheckFun = [&](const boost::system::error_code& error) {
if (error != boost::asio::error::operation_aborted) {
std::lock_guard<std::mutex> _l(_m);
// How to find out whether the client disconnected: Monitor the socket with Posix "select":
// If select says that there is data, available() returns 0 then apparently the client has disconnected.
// WTF, is this really the only way to notice that a client properly disconnected? Internet says so.
auto sockfs = reqImpl->socket().impl()->sockfd();
fd_set fds;
FD_ZERO(&fds);
FD_SET(sockfs, &fds);
struct timeval timeout = {0, 1000};
int sel = select(sockfs + 1, &fds, 0, 0, &timeout);
if (sel < 0 || (sel > 0 && reqImpl->socket().available() == 0)) {
ROS_ERROR_STREAM("Apparently we have a disconnect in device event queue \"" << eventQueue->clientID << "\"");
breakLoop = true;
eventQueue->cancelPopEvent();
}
connCheckTimer.expires_from_now(boost::posix_time::milliseconds(100));
connCheckTimer.async_wait(connCheckFun);
}
};
connCheckTimer.expires_from_now(boost::posix_time::milliseconds(100));
connCheckTimer.async_wait(connCheckFun);
try {
msg = eventQueue->popEvent();
} catch (DeviceEventQueue::TimeoutException& e) {
msg["eventtype"] = "timeoutevent";
}
{
std::lock_guard<std::mutex> _l(_m);
breakLoop = true;
connCheckTimer.cancel();
}
handleResponse(req, resp, msg);
};
void DeviceRepoHandler::__handleEventWebSocket(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp, DeviceEventQueue* eventQueue) {
try {
Net::WebSocket ws(req, resp);
std::cout << "WebSocket connection established." << std::endl;
bool breakLoop = false;
parent->getIOService().post([&]() {
char buffer[1024];
int flags;
int n;
bool timeout = false;
do {
try {
timeout = false;
n = ws.receiveFrame(buffer, sizeof(buffer), flags);
// ToDo: Handle ping and other shit.
std::cout << Poco::format("Frame received (length=%d, flags=0x%x).", n, unsigned(flags)) << std::endl;
} catch (TimeoutException& e) {
// Timeout may happen, nothing fatal.
timeout = true;
} catch (Net::NetException& e) {
breakLoop= true;
ROS_ERROR_STREAM("Exception in WebSocket receive activity: " << e.what());
}
} while (timeout || (!breakLoop && n > 0 && (flags & Net::WebSocket::FRAME_OP_BITMASK) != Net::WebSocket::FRAME_OP_CLOSE));
breakLoop= true;
eventQueue->cancelPopEvent();
});
auto format = getResponseDataFormat(req);
while (!breakLoop) {
Variant msg;
msg = eventQueue->popEvent();
std::string encoded = encodeVariant(msg, req, format);
ws.sendFrame(encoded.c_str(), encoded.size(), Net::WebSocket::FRAME_FLAG_FIN | Net::WebSocket::FRAME_OP_TEXT);
}
std::cout << "WebSocket connection closed." << std::endl;
} catch (Net::WebSocketException& exc) {
std::cout << "WebSocketException caught: " << exc.what() << std::endl;
switch (exc.code()) {
case Net::WebSocket::WS_ERR_HANDSHAKE_UNSUPPORTED_VERSION:
resp.set("Sec-WebSocket-Version", Net::WebSocket::WEBSOCKET_VERSION);
// fallthrough
case Net::WebSocket::WS_ERR_NO_HANDSHAKE:
case Net::WebSocket::WS_ERR_HANDSHAKE_NO_VERSION:
case Net::WebSocket::WS_ERR_HANDSHAKE_NO_KEY:
resp.setStatusAndReason(Net::HTTPResponse::HTTP_BAD_REQUEST);
resp.setContentLength(0);
resp.send();
break;
}
}
};
void DeviceRepoHandler::__handleEvent(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp, bool websocket) {
std::vector<std::string> pathPieces, matchPieces;
@@ -476,9 +375,9 @@ void DeviceRepoHandler::__handleEvent(Poco::Net::HTTPServerRequest &req, Poco::N
}
eventQueue->checkRegistration(event, devptr, path.str());
if (websocket) {
__handleEventWebSocket(req, resp, eventQueue.get());
__handleEventWebSocket(req, resp, eventQueue.get(), parent);
} else {
__handleEventLongPolling(req, resp, eventQueue.get());
__handleEventLongPolling(req, resp, eventQueue.get(), parent);
}
eventQueue->unsetHandler();
} else {
@@ -659,7 +558,9 @@ DeviceAPIMethod DeviceRepoHandler::getDeviceAPIMethod(Net::HTTPServerRequest &re
methodRaw = req.get("X-API-Method");
}
if (!methodRaw.empty()) {
if (methodRaw == "command") {
if (methodRaw == "get") {
retval = DeviceAPIMethod::GET;
} else if (methodRaw == "command") {
retval = DeviceAPIMethod::COMMAND;
} else if (methodRaw == "statusupdate") {
retval = DeviceAPIMethod::STATUSUPDATE;
@@ -121,10 +121,6 @@ public:
void __handleEvent(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp, bool websocket);
void __handleEventLongPolling(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp, DeviceEventQueue* eventQueue);
void __handleEventWebSocket(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp, DeviceEventQueue* eventQueue);
void sendDeviceCommand( devicerepository::Device* device,
Poco::Net::HTTPServerRequest &req,
const std::string& cmd,
+4
View File
@@ -19,6 +19,7 @@ find_package(APR REQUIRED)
find_package(CELIX REQUIRED)
find_package(Boost REQUIRED)
find_package(libvariant REQUIRED)
find_package(Poco REQUIRED COMPONENTS Net NetSSL)
###################################
@@ -44,6 +45,7 @@ include_directories(
${APR_INCLUDE_DIR}
${APR_INCLUDE_DIR}/apr-1.0
${libvariant_INCLUDE_DIR}
${Poco_INCLUDE_DIRS}
)
set(BUNDLE_SYMBOLICNAME "hos.hos_ruleengine")
@@ -60,12 +62,14 @@ bundle(hos_cosgi_ruleengine
${PROJECT_SOURCE_DIR}/src/hos/cosgi/service/ruleengine/rules/conditions/LogicalOrCondition
${PROJECT_SOURCE_DIR}/src/hos/cosgi/service/ruleengine/rules/conditions/LogicalAndCondition
${PROJECT_SOURCE_DIR}/src/hos/cosgi/service/ruleengine/rules/conditions/DeviceMsgCondition
${PROJECT_SOURCE_DIR}/src/hos/cosgi/service/ruleengine/webresource/RuleEngineHandler
)
target_link_libraries(hos_cosgi_ruleengine
${catkin_LIBRARIES}
${APR_LIBRARIES}
${libvariant_LIBRARY}
${Poco_LIBRARIES}
)
@@ -68,11 +68,16 @@ enum class RuleEngineEventType : int {
RULE_REMOVED,
RULE_UPDATED,
RULE_STATE_CHANGED,
RULECONDITION_ADDED,
RULECONDITION_REMOVED,
RULECONDITION_UPDATED,
RULECONDITION_STATE_CHANGED,
ALL_EVENTS,
CHANGE_EVENTS
};
class RuleEngineEvent {
public:
typedef std::chrono::system_clock::time_point timestamp_t;
@@ -86,6 +91,7 @@ public:
};
/**
* Public COSGI interface of the rule engine service.
*/
@@ -3,6 +3,7 @@
#include <hos/cosgi/service/Interface.h>
#include <hos/cosgi/service/ruleengine/RuleEngineService.h>
#include <Variant/Variant.h>
@@ -14,6 +15,7 @@ namespace service {
// forward declarations
namespace devicerepository { class DeviceRepositoryService; }
namespace scripting { class Script; }
// main namespace
@@ -34,7 +36,7 @@ class RulePackage;
* Public COSGI interface of a rule.
*/
class Rule
: public virtual interface::Object {
: public virtual interface::ObservableObject<RuleEngineEventType, RuleEngineEvent> {
public:
@@ -91,6 +93,10 @@ public:
*/
virtual void executeExitScript(const libvariant::Variant& lContext = libvariant::Variant()) = 0;
virtual std::shared_ptr<scripting::Script> getScript() = 0;
virtual std::shared_ptr<scripting::Script> getExitScript() = 0;
/**
* Returns whether this rule is enabled.
*
@@ -3,6 +3,7 @@
#include <hos/cosgi/service/Interface.h>
#include <hos/cosgi/service/ruleengine/RuleEngineService.h>
#include <vector>
#include <memory>
#include <string>
@@ -40,7 +41,7 @@ class Rule;
* Public COSGI interface of a rule package.
*/
class RulePackage
: public virtual interface::Object {
: public virtual interface::ObservableObject<RuleEngineEventType, RuleEngineEvent> {
public:
@@ -3,6 +3,7 @@
#include <hos/cosgi/service/Interface.h>
#include <hos/cosgi/service/ruleengine/RuleEngineService.h>
#include <memory>
#include <vector>
#include <Variant/Variant.h>
@@ -50,7 +51,7 @@ enum class ConditionTriggerType : int {
* Public COSGI interface of a rule condition.
*/
class RuleCondition
: public virtual interface::Object {
: public virtual interface::ObservableObject<RuleEngineEventType, RuleEngineEvent> {
public:
@@ -4,8 +4,29 @@
#include "RuleCondition.h"
#include <hos/cosgi/service/Implementation.h>
#include <hos/cosgi/service/core/CoreService.h>
// Quick and dirty workaround
// ToDo: Find better solution
#ifndef _STD_HASH_RULEENGINEEVENTTYPE_
#define _STD_HASH_RULEENGINEEVENTTYPE_
namespace std {
/**
* Specialization of std::hash for RuleEngineEventType.
*
* Required by std::map.
*/
template <>
class hash <hos::cosgi::service::ruleengine::RuleEngineEventType>{
public :
size_t operator()(const hos::cosgi::service::ruleengine::RuleEngineEventType& x) const {
return static_cast<size_t>(x);
}
};
} // namespace std
#endif
// main namespace
namespace hos {
@@ -22,7 +43,7 @@ namespace conditions {
*/
class RuleConditionImpl
: public virtual RuleCondition,
public implementation::Object {
public implementation::ObservableObject<RuleEngineEventType, RuleEngineEvent> {
protected:
@@ -13,6 +13,7 @@
#include <hos/cosgi/celix.h>
// CoreService
#include "hos/cosgi/service/ruleengine/RuleEngineServiceImpl.h"
#include "hos/cosgi/service/ruleengine/webresource/RuleEngineHandler.h"
#include <hos/cosgi/service/core/CoreService.h>
#include <hos/cosgi/service/devicerepository/DeviceRepositoryService.h>
@@ -23,13 +24,16 @@ using namespace hos::cosgi::service::scripting;
using namespace hos::cosgi::service::devicerepository;
using namespace hos::cosgi::service::core;
using namespace hos::cosgi;
namespace ruleweb = hos::cosgi::service::ruleengine::webresource;
struct UserData {
apr_pool_t* memoryPool = nullptr;
std::unique_ptr<RuleEngineServiceImpl> serviceImpl = nullptr;
std::unique_ptr<RuleEngineServiceImpl> serviceImpl;
std::unique_ptr<ruleweb::RuleEngineHandlerFactory> webResourceImpl;
service_registration_pt serviceReg = nullptr;
service_registration_pt webResourceReg = nullptr;
serviceListener myServiceListener;
bundle_context_pt bundleContext;
};
@@ -56,10 +60,16 @@ celix_status_t cb_serviceChanged(void* listener, service_event_pt event) {
checkReg = true;
}
// Register rule engine service
// Register rule engine service and webresource
if (checkReg && !myUserData->serviceReg && myUserData->serviceImpl->isInitialized()) {
RuleEngineService* engineInterface = dynamic_cast<RuleEngineService*>(myUserData->serviceImpl.get());
bundleContext_registerService(myUserData->bundleContext, (char*)RuleEngineServiceImpl::SERVICE_NAME, engineInterface, nullptr, &myUserData->serviceReg);
// web resource
service::webserver::WebResourceHandlerFactory* webInterface = dynamic_cast<service::webserver::WebResourceHandlerFactory*>(myUserData->webResourceImpl.get());
// ToDo: Properly delete properties
properties_pt webResourceProps = properties_create();
properties_set(webResourceProps, (char*)"path", (char*)"/ruleengine0");
bundleContext_registerService(myUserData->bundleContext, (char*)service::webserver::WebResourceHandlerFactory::SERVICE_NAME, webInterface, webResourceProps, &myUserData->webResourceReg);
}
} else if (event->type == OSGI_FRAMEWORK_SERVICE_EVENT_UNREGISTERING) {
@@ -110,6 +120,12 @@ celix_status_t bundleActivator_start(void * userData, bundle_context_pt context)
}
myUserData->serviceImpl.reset(new RuleEngineServiceImpl());
// Create rule engine webresource
if (myUserData->webResourceReg != nullptr) {
serviceRegistration_unregister(myUserData->webResourceReg);
}
myUserData->webResourceImpl.reset(new ruleweb::RuleEngineHandlerFactory(myUserData->serviceImpl.get()));
// Get core Service
service_reference_pt ref = nullptr;
bundleContext_getServiceReference(context, (char*)CoreService::SERVICE_NAME, &ref);
@@ -130,10 +146,16 @@ celix_status_t bundleActivator_start(void * userData, bundle_context_pt context)
myUserData->serviceImpl->setRepoService(repoService);
}
// Register rule engine service
// Register rule engine service and webresource
if (myUserData->serviceImpl->isInitialized()) {
RuleEngineService* engineInterface = dynamic_cast<RuleEngineService*>(myUserData->serviceImpl.get());
bundleContext_registerService(context, (char*)RuleEngineServiceImpl::SERVICE_NAME, engineInterface, nullptr, &myUserData->serviceReg);
// web resource
service::webserver::WebResourceHandlerFactory* webInterface = dynamic_cast<service::webserver::WebResourceHandlerFactory*>(myUserData->webResourceImpl.get());
// ToDo: Properly delete properties
properties_pt webResourceProps = properties_create();
properties_set(webResourceProps, (char*)"path", (char*)"/ruleengine0");
bundleContext_registerService(context, (char*)service::webserver::WebResourceHandlerFactory::SERVICE_NAME, webInterface, webResourceProps, &myUserData->webResourceReg);
}
// Register service listener
@@ -155,6 +177,13 @@ celix_status_t bundleActivator_stop(void * userData, bundle_context_pt context)
if (userData != nullptr) {
UserData* myUserData = (UserData*)userData;
status = bundleContext_removeServiceListener(context, &myUserData->myServiceListener);
if (myUserData->webResourceReg != nullptr) {
serviceRegistration_unregister(myUserData->webResourceReg);
myUserData->webResourceReg = nullptr;
}
if (myUserData->webResourceImpl) {
myUserData->webResourceImpl.reset();
}
if (myUserData->serviceReg != nullptr) {
serviceRegistration_unregister(myUserData->serviceReg);
myUserData->serviceReg = nullptr;
@@ -21,7 +21,10 @@
#include <memory>
#include <string>
// Quick and dirty workaround
// ToDo: Find better solution
#ifndef _STD_HASH_RULEENGINEEVENTTYPE_
#define _STD_HASH_RULEENGINEEVENTTYPE_
namespace std {
/**
* Specialization of std::hash for RuleEngineEventType.
@@ -36,6 +39,7 @@ public :
}
};
} // namespace std
#endif
// Service Namespace
@@ -6,6 +6,7 @@
#include <hos/cosgi/service/ruleengine/rules/conditions/RuleCondition.h>
#include <hos/cosgi/service/devicerepository/DeviceRepositoryService.h>
#include <hos/cosgi/service/scripting/ScriptingService.h>
#include <hos/cosgi/service/Implementation.h>
#include "RulePackageImpl.h"
#include <Variant/Variant.h>
#include <string>
@@ -31,7 +32,7 @@ using namespace libvariant;
*/
class RuleImpl
: public virtual Rule,
public implementation::Object {
public implementation::ObservableObject<RuleEngineEventType, RuleEngineEvent> {
private:
@@ -74,6 +75,14 @@ public:
virtual void executeExitScript(const libvariant::Variant& lContext = libvariant::Variant()) override;
virtual std::shared_ptr<scripting::Script> getScript() override {
return script;
}
virtual std::shared_ptr<scripting::Script> getExitScript() override {
return exitScript;
}
virtual bool isEnabled() noexcept override {
return enabled;
}
@@ -4,6 +4,7 @@
#include <hos/cosgi/service/ruleengine/rules/RulePackage.h>
#include <hos/cosgi/service/scripting/ScriptingService.h>
#include <hos/cosgi/service/Implementation.h>
#include "../RuleEngineServiceImpl.h"
#include <Variant/Variant.h>
#include <vector>
@@ -33,7 +34,7 @@ using namespace ::libvariant;
*/
class RulePackageImpl
: public virtual RulePackage,
public implementation::Object {
public implementation::ObservableObject<RuleEngineEventType, RuleEngineEvent> {
private:
@@ -1,10 +1,328 @@
/*
* RuleEngineHandler.cpp
*
* Created on: 05.06.2015
* Author: matthias
*/
#include "RuleEngineHandler.h"
#include <hos/cosgi/service/ruleengine/rules/RulePackage.h>
#include <hos/cosgi/service/ruleengine/rules/Rule.h>
#include <hos/cosgi/service/scripting/ScriptingService.h>
#include <hos/utils/Timestamp.h>
#include <boost/format.hpp>
#include <boost/algorithm/string.hpp>
#include <Poco/URI.h>
#include <Poco/Timestamp.h>
#include <boost/algorithm/string.hpp>
#include <chrono>
#include <Poco/Net/HTTPServerRequestImpl.h>
#include <Poco/Net/HTTPServerResponseImpl.h>
#include <Poco/Net/WebSocket.h>
#include "Poco/Net/NetException.h"
#include <Poco/Net/StreamSocket.h>
// Service Namespace
namespace hos {
namespace cosgi {
namespace service {
namespace ruleengine {
namespace webresource {
using namespace Poco;
using namespace libvariant;
HOS_SERVICE_IMPL_INFO( RuleEngineHandlerFactory, // Implementation
service::webserver::WebResourceHandlerFactory::SERVICE_NAME, // Service Name
1, 0 ) // Version
RuleEngineHandlerFactory::RuleEngineHandlerFactory(RuleEngineService* ruleService)
: ruleService(ruleService),
pkgInfoFuncs([](const std::string& key)->pkgInfoFuncs_t::Func_t {
throw webserver::webresource::WebResourceHandlerException((boost::format("Unknown category \"%1%\"") % key).str(),
Poco::Net::HTTPResponse::HTTPStatus::HTTP_NOT_FOUND);
}),
ruleInfoFuncs([](const std::string& key)->ruleInfoFuncs_t::Func_t {
throw webserver::webresource::WebResourceHandlerException((boost::format("Unknown category \"%1%\"") % key).str(),
Poco::Net::HTTPResponse::HTTPStatus::HTTP_NOT_FOUND);
}) {
pkgInfoFuncs.add("id", [](rules::RulePackage* pkg, const std::string& category)->Variant {
return pkg->getName();
});
pkgInfoFuncs.add("language", [](rules::RulePackage* pkg, const std::string& category)->Variant {
return pkg->getLanguage();
});
pkgInfoFuncs.add("enabled", [](rules::RulePackage* pkg, const std::string& category)->Variant {
return pkg->isEnabled();
});
pkgInfoFuncs.add("globalcontext", [](rules::RulePackage* pkg, const std::string& category)->Variant {
auto gc = pkg->getGlobalScriptContext();
if (!gc.expired()) {
return gc.lock()->getJSONRepresentation();
} else {
return libvariant::Variant::NullType;
}
});
pkgInfoFuncs.add("rules", [this](rules::RulePackage* pkg, const std::string& category)->Variant {
Variant retval = libvariant::Variant::List();
for (auto r : pkg->getRules()) {
if (!r.expired()) {
retval.Append(getRuleInfo(r.lock().get()));
}
}
return retval;
});
ruleInfoFuncs.add("id", [](rules::Rule* rule, const std::string& category)->Variant {
return rule->getName();
});
ruleInfoFuncs.add("enabled", [](rules::Rule* rule, const std::string& category)->Variant {
return rule->isEnabled();
});
ruleInfoFuncs.add("script", [](rules::Rule* rule, const std::string& category)->Variant {
auto script = rule->getScript();
if (script) {
return script->getJSONRepresentation();
} else {
return Variant::NullType;
}
});
ruleInfoFuncs.add("exitscript", [](rules::Rule* rule, const std::string& category)->Variant {
auto script = rule->getExitScript();
if (script) {
return script->getJSONRepresentation();
} else {
return Variant::NullType;
}
});
ruleInfoFuncs.add("condition", [](rules::Rule* rule, const std::string& category)->Variant {
return "ToDo";
});
}
Net::HTTPRequestHandler* RuleEngineHandlerFactory::createRequestHandler( Poco::URI matchUri, const Net::HTTPServerRequest& request, bool ssl ) noexcept {
return new RuleEngineHandler( this, matchUri, ruleService, ssl);
}
Variant RuleEngineHandlerFactory::getRulePkgInfo( rules::RulePackage* pkg, const std::string& category, const std::string& path ) {
Variant retval;
// We want all info categories
if (category.empty()) {
pkgInfoFuncs.forEach([&](const std::string& key, const Variant& value){
retval[key] = value;
}, pkg, category);
// We want a single info category
} else {
retval = pkgInfoFuncs.call(category, pkg, category);
if (!path.empty()) {
if (retval.HasPath(path)) {
retval = retval.GetPath(path);
} else {
throw webserver::webresource::WebResourceHandlerException((boost::format("Unknown path \"%1%\" for rule package info category \"%2%\".") % path % category).str(),
Poco::Net::HTTPResponse::HTTPStatus::HTTP_NOT_FOUND);
}
}
}
return retval;
}
Variant RuleEngineHandlerFactory::getRuleInfo( rules::Rule* rule, const std::string& category, const std::string& path ) {
Variant retval;
// We want all info categories
if (category.empty()) {
ruleInfoFuncs.forEach([&](const std::string& key, const Variant& value){
retval[key] = value;
}, rule, category);
// We want a single info category
} else {
retval = ruleInfoFuncs.call(category, rule, category);
if (!path.empty()) {
if (retval.HasPath(path)) {
retval = retval.GetPath(path);
} else {
throw webserver::webresource::WebResourceHandlerException((boost::format("Unknown path \"%1%\" for rule info category \"%2%\".") % path % category).str(),
Poco::Net::HTTPResponse::HTTPStatus::HTTP_NOT_FOUND);
}
}
}
return retval;
}
void RuleEngineHandler::_handleRequest(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp) {
if(req.find("Upgrade") != req.end() && Poco::icompare(req["Upgrade"], "websocket") == 0) {
//__handleEvent(req, resp, true);
/*throw webserver::webresource::WebResourceHandlerException("ToDo",
Net::HTTPResponse::HTTP_NOT_IMPLEMENTED);*/
} else {
if (req.getMethod() == Net::HTTPRequest::HTTP_PUT) {
// Put Command
throw webserver::webresource::WebResourceHandlerException("Still a ToDo.",
Net::HTTPResponse::HTTP_NOT_IMPLEMENTED);
} else if (req.getMethod() == Net::HTTPRequest::HTTP_GET || req.getMethod() == Net::HTTPRequest::HTTP_POST) {
RuleEngineAPIMethod method = getRuleEngineAPIMethod(req);
if (method == RuleEngineAPIMethod::GET) {
__handleInfo(req, resp);
} else if (method == RuleEngineAPIMethod::SET) {
//__handleCommand(req, resp, method);
} else if (method == RuleEngineAPIMethod::EVENT) {
//__handleEvent(req, resp, false);
} else {
throw webserver::webresource::WebResourceHandlerException("Could not understood request",
Net::HTTPResponse::HTTP_BAD_REQUEST);
}
} else {
throw webserver::webresource::WebResourceHandlerException("Could not understood request",
Net::HTTPResponse::HTTP_BAD_REQUEST);
}
}
}
void RuleEngineHandler::__handleInfo(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp) {
std::vector<std::string> pathPieces, matchPieces;
matchUri.getPathSegments(matchPieces);
Poco::URI uri(req.getURI());
uri.getPathSegments(pathPieces);
std::unique_ptr<Net::HTMLForm> form;
if (req.getMethod() == Net::HTTPRequest::HTTP_POST) {
form.reset(new Net::HTMLForm(req, req.stream()));
} else {
form.reset(new Net::HTMLForm(req));
}
// Nothing given, query mode (there is no real query functionality implemented yet though)
if (pathPieces.size() <= matchPieces.size()) {
auto pkgs = ruleService->getRulePackages();
// Parse HTTP header "Range" parameter (to support dojo/store)
size_t rangeStart = 0;
size_t rangeStop = pkgs.size() - 1;
std::string rangeRaw;
if (req.has("Range")) {
rangeRaw = req.get("Range");
} else if (req.has("X-Range")) {
rangeRaw = req.get("X-Range");
}
if (!rangeRaw.empty()) {
if (boost::algorithm::starts_with(rangeRaw, "items=")) {
auto range = rangeRaw.substr(6);
std::vector<std::string> splitParts;
boost::algorithm::split(splitParts, range, boost::algorithm::is_any_of("-"));
if (splitParts.size() == 2) {
try {
size_t r1 = std::stol(splitParts[0]);
size_t r2 = std::stol(splitParts[1]);
if (r1 <= r2) {
rangeStart = r1;
rangeStop = std::min(r2, rangeStop);
}
} catch (...) {
ROS_ERROR_STREAM("Could not parse Range parameter \"" << rangeRaw << "\".");
}
}
}
}
// Iterate over packages
Variant result = Variant::List();
if (pkgs.size() > 0) {
for (size_t t = rangeStart; t <= rangeStop; t++) {
auto pkg = pkgs[t].lock();
result.Append(parent->getRulePkgInfo(pkg.get()));
}
}
// There are no devices and a range was given, probably out of range
if (result.Size() == 0 && rangeStart > 0) {
if (rangeStart > 0) {
handleErrorResponse(req, resp, "Out of range", Poco::Net::HTTPResponse::HTTP_NOT_FOUND);
} else {
handleResponse(req, resp, result);
}
// Return devices array
} else {
std::string rangeOut = (boost::format("items %1%-%2%/%3%") % rangeStart % rangeStop % pkgs.size()).str();
resp.set("Content-Range", rangeOut);
resp.set("X-Content-Range", rangeOut);
handleResponse(req, resp, result);
}
// rule package given
} else if (pathPieces.size() == matchPieces.size() + 1) {
// rule given
} else if (pathPieces.size() == matchPieces.size() + 2) {
}
}
RuleEngineAPIMethod RuleEngineHandler::getRuleEngineAPIMethod(Net::HTTPServerRequest &req) noexcept {
std::unique_ptr<Net::HTMLForm> form;
RuleEngineAPIMethod retval = RuleEngineAPIMethod::GET; // default
if (req.getMethod() == Net::HTTPRequest::HTTP_POST && req.has("Content-type")
&& req.get("Content-type") == Net::HTMLForm::ENCODING_URL) {
form.reset(new Net::HTMLForm(req, req.stream()));
} else {
form.reset(new Net::HTMLForm(req));
}
std::string methodRaw;
if (form->has("method")) {
methodRaw = form->get("method");
} else if (req.has("API-Method")) {
methodRaw = req.get("API-Method");
} else if (req.has("X-API-Method")) {
methodRaw = req.get("X-API-Method");
}
if (!methodRaw.empty()) {
if (methodRaw == "get") {
retval = RuleEngineAPIMethod::GET;
} else if (methodRaw == "set") {
retval = RuleEngineAPIMethod::SET;
} else if (methodRaw == "event") {
retval = RuleEngineAPIMethod::EVENT;
} else {
throw webserver::webresource::WebResourceHandlerException(
(boost::format("Unknown rule engine api method \"%1%\"") % methodRaw).str(),
Net::HTTPResponse::HTTPStatus::HTTP_BAD_REQUEST );
}
} else if (form->has("onevent") || req.has("On-Event") || req.has("X-On-Event")) {
retval = RuleEngineAPIMethod::EVENT;
}
return retval;
}
} // namespace webresource
} // namespace ruleengine
} // namespace service
} // namespace cosgi
} // namespace hos
@@ -6,6 +6,7 @@
#include <hos/cosgi/service/webserver/WebServerService.h>
#include <hos/cosgi/service/webserver/webresource/WebResourceHandlerImpl.h>
#include <hos/cosgi/service/ruleengine/RuleEngineService.h>
#include <hos/utils/FunctionTable.h>
#include <Poco/Net/HTTPRequestHandler.h>
#include <Variant/Variant.h>
#include <ros/ros.h>
@@ -42,33 +43,59 @@ enum class RuleEngineAPIMethod {
typedef webserver::webresource::WebResourceEventQueue<
interface::Observer<RuleEngineEventType, RuleEngineEvent>,
RuleEngineHandlerFactory,
RuleEngineHandler
> RuleEngineEventQueueBase;
class RuleEngineEventQueue
: public RuleEngineEventQueueBase {
public:
RuleEngineEventType eventType = RuleEngineEventType::ALL_EVENTS;
RuleEngineService* const ruleService;
RuleEngineEventQueue( const std::string& clientID,
RuleEngineHandlerFactory* handlerFactory,
RuleEngineService* ruleService )
: WebResourceEventQueue(clientID, handlerFactory), ruleService(ruleService) {}
virtual ~RuleEngineEventQueue() {
unregisterEvents();
}
/*void checkRegistration(RuleEngineEventType eventType, Device* device, const std::string& pointer);
void registerEvents(RuleEngineEventType eventType, Device* device = nullptr);*/
void unregisterEvents() {
}
virtual void notificationReceived( interface::Observer<RuleEngineEventType, RuleEngineEvent>::observable_type* o,
const RuleEngineEventType& key,
const RuleEngineEvent* e) noexcept override {
}
};
class RuleEngineHandlerFactory
: public virtual webserver::WebResourceHandlerFactory,
public implementation::Service<RuleEngineHandlerFactory> {
: public webserver::webresource::WebResourceHandlerFactoryImpl<RuleEngineHandlerFactory, false, RuleEngineEventQueue> {
public:
typedef std::function<libvariant::Variant( rules::RulePackage* pkg, const std::string& category )> pkgInfoFunc_t;
typedef std::function<libvariant::Variant( rules::Rule* pkg, const std::string& category )> ruleInfoFuncFunc_t;
private:
webserver::WebServerService* service = nullptr;
typedef utils::FunctionTable<std::string, libvariant::Variant, rules::RulePackage*, const std::string&> pkgInfoFuncs_t;
typedef utils::FunctionTable<std::string, libvariant::Variant, rules::Rule*, const std::string&> ruleInfoFuncs_t;
RuleEngineService* ruleService;
boost::asio::io_service io_service;
std::unique_ptr<std::thread> io_service_thread;
pkgInfoFuncs_t pkgInfoFuncs;
ruleInfoFuncs_t ruleInfoFuncs;
public:
RuleEngineHandlerFactory(RuleEngineService* ruleService);
~RuleEngineHandlerFactory();
virtual Poco::Net::HTTPRequestHandler* createRequestHandler(Poco::URI matchUri, const Poco::Net::HTTPServerRequest& request, bool ssl ) noexcept override;
virtual void setWebServerService(webserver::WebServerService* service) noexcept override {
this->service = service;
}
libvariant::Variant getRulePkgInfo( rules::RulePackage* pkg, const std::string& category = "", const std::string& path = "" );
virtual bool excludeFromAuthentication() noexcept override {
return false;
}
boost::asio::io_service& getIOService() {
return io_service;
}
libvariant::Variant getRuleInfo( rules::Rule* pkg, const std::string& category = "", const std::string& path = "" );
};
@@ -82,6 +109,7 @@ protected:
RuleEngineService* ruleService;
bool ssl;
public:
RuleEngineHandler( RuleEngineHandlerFactory* parent,
@@ -92,11 +120,23 @@ public:
virtual void _handleRequest(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp) override;
void __handleInfo(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp);
void __handleSet(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp, RuleEngineAPIMethod method);
void __handleEvent(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp, bool websocket);
static RuleEngineAPIMethod getRuleEngineAPIMethod(Poco::Net::HTTPServerRequest &req) noexcept;
/*void sendDeviceCommand( devicerepository::Device* device,
Poco::Net::HTTPServerRequest &req,
const std::string& cmd,
const std::vector<std::string>& path = std::vector<std::string>() );*/
};
} // namespace webresource
} // namespace ruleengine
} // namespace service
@@ -25,7 +25,10 @@ namespace scripting {
* Currently it is just an opaque handle that is passed to the script execution handle.
*/
class GlobalScriptContext
: public virtual interface::Object {};
: public virtual interface::Object {
public:
virtual libvariant::Variant getJSONRepresentation() = 0;
};
@@ -49,6 +52,8 @@ public:
*/
virtual void execute( std::shared_ptr<GlobalScriptContext> gContext = std::shared_ptr<GlobalScriptContext>(),
const libvariant::Variant& lContext = libvariant::Variant() ) = 0;
virtual libvariant::Variant getJSONRepresentation() = 0;
};
@@ -53,6 +53,16 @@ public:
PyGlobalScriptContext(PyScriptingService* service, const boost::python::object& gContext, const std::string& code)
: service(service), gContext(gContext), code(code) {}
/* Public Service Implementations */
virtual libvariant::Variant getJSONRepresentation() override {
libvariant::Variant retval;
retval["code"] = code;
return retval;
}
/* Internal */
const std::string& getScriptCode() noexcept {
@@ -90,6 +100,12 @@ public:
virtual void execute( std::shared_ptr<GlobalScriptContext> gContext = std::shared_ptr<GlobalScriptContext>(),
const libvariant::Variant& lContext = libvariant::Variant() ) override;
virtual libvariant::Variant getJSONRepresentation() override {
libvariant::Variant retval;
retval["code"] = code;
return retval;
}
/* Internal */
@@ -1,23 +1,32 @@
define([
"dojo/_base/config",
"dojo/store/JsonRest",
'dojo/store/Observable',
'dojo/_base/declare',
'dstore/Rest',
'dstore/RequestMemory',
'dstore/SimpleQuery',
'dstore/Trackable',
"dojox/socket",
"dojo/_base/xhr"
], function(config, JsonRest, Observable, Socket, xhr){
var _store = new Observable(new JsonRest({
"dojo/_base/xhr",
"dojox/uuid/generateRandomUuid"
], function(config, declare, Rest, RequestMemory, SimpleQuery, Trackable, Socket, xhr, generateRandomUuid){
var RuleEngineStore = declare([Rest, RequestMemory, SimpleQuery, Trackable]);
var _store = new RuleEngineStore({
target: config.hos.deviceRepoUrl
}));
});
var _updateSocketOpened = false;
var _updateSocketClientId = generateRandomUuid();
var _updateSocket = new Socket({
url: config.hos.deviceRepoUrl + '?onevent=changes&clientid=1234',
headers: {
"On-Event": "changes",
"Connection-Timeout": 60000,
"Client-Id": 1234
},
transport: xhr.post });
_updateSocket = Observable(_updateSocket);
url: config.hos.deviceRepoUrl + '?onevent=changes&clientid=' + _updateSocketClientId,
headers: {
"On-Event": "changes",
"Connection-Timeout": 60000,
"Client-Id": 1234
},
transport: xhr.post
});
_updateSocket.on("open", function(event){
_socketOpened = true;
console.log("Socket Opened ", event.message);
@@ -33,9 +42,10 @@ define([
_updateSocket.on("message", function(event) {
var data = JSON.parse(event.data);
switch (data['eventtype']) {
case "statechanged":
_store.notify(data['device'], data['deviceid']);
break;
case "statechanged":
_store.evict(data['deviceid']);
_store.emit('update', {id: data['deviceid'], target: data['device']});
break;
}
console.log("Message: " + data);
});
@@ -51,5 +61,6 @@ define([
// browser supports XHR2
});
};
return _store;
});
});
@@ -0,0 +1,21 @@
define([
"dojo/_base/config",
'dojo/_base/declare',
'dstore/Rest',
'dstore/RequestMemory',
'dstore/SimpleQuery',
'dstore/Trackable',
'dstore/Tree',
"dojox/socket",
"dojo/_base/xhr",
"dojox/uuid/generateRandomUuid"
], function(config, declare, Rest, RequestMemory, SimpleQuery, Trackable, Tree, Socket, xhr, generateRandomUuid){
var RuleEngineStore = declare([Rest, RequestMemory, SimpleQuery, Trackable, Tree]);
var _store = new RuleEngineStore({
target: config.hos.ruleEngineUrl
});
return _store;
});
@@ -0,0 +1,52 @@
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>HOS Widget Test</title>
<link rel="stylesheet" href="libs/dijit/themes/claro/claro.css" media="screen">
<!-- configure Dojo -->
<script>
var dojoConfig = {
async: true,
parseOnLoad: true,
isDebug: true,
deps: [
"dojo/parser"
],
hos: {
ruleEngineUrl: "/hos/api/ruleengine0/"
}
};
</script>
<!-- load Dojo -->
<script src="libs/dojo/dojo.js"
data-dojo-config="async: true"></script>
</head>
<body class="claro">
<h1 id="greeting">HOS Rules Test</h1>
<li id="ruleslist"></li>
<script>
var rules;
require([
"hos/RuleEngine",
"dojo/dom-construct",
"dojo/dom",
"dojo/on",
"dojo/domReady!"
], function(RuleEngine, domConstruct, dom, on){
var test1 = RuleEngine.get("myid/bla");
console.log("test1: ", test1);
var test2 = RuleEngine.filter({arg01: "param01"});
console.log("test2: ", test2);
});
</script>
</body>
</html>
@@ -0,0 +1,5 @@
.claro .dijitToggleButtonChecked .dijitButtonNode {
background-color: #0090ff;
box-shadow: 0 1px 1px rgba(0, 0, 0, 0.2) inset;
transition-duration: 0.1s;
}
@@ -0,0 +1,122 @@
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>Tutorial: Hello Dojo!</title>
<link rel="stylesheet" href="libs/dijit/themes/claro/claro.css" media="screen">
<link rel="stylesheet" href="test04.css" media="screen">
<!-- configure Dojo -->
<script>
var dojoConfig = {
async: true,
parseOnLoad: true,
isDebug: true,
deps: [
"dojo/parser"
],
hos: {
deviceRepoUrl: "/hos/api/devices/"
}
};
</script>
<!-- load Dojo -->
<script src="libs/dojo/dojo.js"
data-dojo-config="async: true"></script>
</head>
<body class="claro">
<h1 id="greeting">Switches</h1>
<li id="devicelist"></li>
<script>
var devices;
var widgets = {};
require([
"hos/DeviceRepo",
"dojo/dom-construct",
"dojo/dom",
"dojo/on",
"dijit/form/ToggleButton",
"dijit/form/HorizontalSlider",
"dojo/domReady!"
], function(DeviceRepo, domConstruct, dom, on, ToggleButton, HorizontalSlider){
var self = this;
devices = DeviceRepo.query();
devices.then(function(data){
console.log("Data: ", data);
for (var i = 0; i < data.length; i++) {
console.log("Device " + i + ": ", data[i]);
if (data[i].type == "core/devices/switch") {
var node = domConstruct.toDom("<ul><button id=\"" + data[i].id + "\"></button></ul>");
domConstruct.place(node, "devicelist");
var b = new ToggleButton({
showLabel: true,
checked: false,
label: data[i].id,
iconClass: 'dijitCheckBoxIcon'
}, data[i].id);
b.startup();
function addListener(id, b) {
on(b, "click", function(e){
DeviceRepo.sendCommand(id, {switch:{set:b.get("checked")}});
});
};
addListener(data[i].id, b);
b.set('checked', data[i].current_state.switch.value);
widgets[data[i].id] = b;
} else if (data[i].type == "core/devices/dimmer") {
var node = domConstruct.toDom("<ul><div id=\"" + data[i].id + "\"></div></ul>");
domConstruct.place(node, "devicelist");
var s = new HorizontalSlider({
name: data[i].id,
value: 0,
minimum: 0,
maximum: 1,
intermediateChanges: true,
style: "width:300px;"
}, data[i].id);
s.startup();
function addListener(id, s) {
on(s, "click", function(e){
DeviceRepo.sendCommand(id, {dimmer:{set:s.get("value")}});
});
};
addListener(data[i].id, s);
s.set('value', data[i].current_state.dimmer.value);
widgets[data[i].id] = s;
}
}
}, function(err){
console.log("Error: ", err);
}, function(update){
console.log("Update: ", update);
}
);
devices.observe(function(device, removedFrom, insertedInto){
// this will be called any time a market is added, removed, or updated
console.log("Observe triggered: ", device, ', ', removedFrom, ', ', insertedInto);
if (removedFrom == insertedInto) {
if (device.type == "core/devices/switch") {
var b = widgets[device.id];
b.set('checked', device.current_state.switch.value);
} else if (device.type == "core/devices/dimmer") {
var s = widgets[device.id];
s.set('value', device.current_state.dimmer.value);
}
}
}, true); // we can indicate to be notified of object updates as well
});
</script>
</body>
</html>
@@ -50,7 +50,8 @@
var self = this;
devices = DeviceRepo.query();
_devices = DeviceRepo.filter().track();
devices = _devices.fetch();
devices.then(function(data){
console.log("Data: ", data);
for (var i = 0; i < data.length; i++) {
@@ -101,19 +102,18 @@
console.log("Update: ", update);
}
);
devices.observe(function(device, removedFrom, insertedInto){
_devices.on('update', function(event){
// this will be called any time a market is added, removed, or updated
console.log("Observe triggered: ", device, ', ', removedFrom, ', ', insertedInto);
if (removedFrom == insertedInto) {
if (device.type == "core/devices/switch") {
var b = widgets[device.id];
b.set('checked', device.current_state.switch.value);
} else if (device.type == "core/devices/dimmer") {
var s = widgets[device.id];
s.set('value', device.current_state.dimmer.value);
}
}
}, true); // we can indicate to be notified of object updates as well
console.log("Observe triggered: ", event);
var device = event.target;
if (device.type == "core/devices/switch") {
var b = widgets[device.id];
b.set('checked', device.current_state.switch.value);
} else if (device.type == "core/devices/dimmer") {
var s = widgets[device.id];
s.set('value', device.current_state.dimmer.value);
}
}); // we can indicate to be notified of object updates as well
});
</script>
@@ -21,6 +21,16 @@
#include <mutex>
#include <condition_variable>
#include <Poco/Net/HTTPServerRequestImpl.h>
#include <Poco/Net/HTTPServerResponseImpl.h>
#include <Poco/Net/WebSocket.h>
#include "Poco/Net/NetException.h"
#include <Poco/Net/StreamSocket.h>
#include <sys/socket.h>
#include <sys/select.h>
#include <sys/time.h>
#include <stropts.h>
// service namespace
@@ -279,7 +289,7 @@ public:
enum class ResponseDataFormat : int {
enum class DataFormat : int {
JSON,
JSONP,
YAML,
@@ -298,47 +308,87 @@ protected:
class NotImplementedException
: public std::exception {};
ResponseDataFormat getResponseDataFormat(const Poco::Net::HTTPServerRequest& request) {
ResponseDataFormat retval = ResponseDataFormat::JSON;
DataFormat getDataFormat(const Poco::Net::HTTPServerRequest& request, bool incoming = false) {
DataFormat retval = DataFormat::JSON; // defaults to json
Poco::Net::HTMLForm form(request);
if (form.has("format")) {
std::string formatRaw = form.get("format");
if (formatRaw.compare("json") == 0) {
// do nothing
retval = DataFormat::JSON;
} else if (formatRaw.compare("jsonp") == 0) {
retval = ResponseDataFormat::JSONP;
retval = DataFormat::JSONP;
} else if (formatRaw.compare("yaml") == 0) {
retval = ResponseDataFormat::YAML;
retval = DataFormat::YAML;
} else if (formatRaw.compare("xmlplist") == 0) {
retval = ResponseDataFormat::XMLPLIST;
retval = DataFormat::XMLPLIST;
} else if (formatRaw.compare("bundlehdr") == 0) {
retval = ResponseDataFormat::BUNDLEHDR;
retval = DataFormat::BUNDLEHDR;
} else if (formatRaw.compare("msgpack") == 0) {
retval = ResponseDataFormat::MSGPACK;
retval = DataFormat::MSGPACK;
} else {
throw WebResourceHandlerException((boost::format("Unknown format parameter \"%1%\"") % formatRaw).str(),
Poco::Net::HTTPResponse::HTTPStatus::HTTP_BAD_REQUEST);
}
} else if (form.has("json")) {
retval = DataFormat::JSON;
} else if (form.has("jsonp")) {
retval = DataFormat::JSONP;
} else if (form.has("yaml")) {
retval = DataFormat::YAML;
} else if (form.has("xmlplist")) {
retval = DataFormat::XMLPLIST;
} else if (form.has("bundlehdr")) {
retval = DataFormat::BUNDLEHDR;
} else if (form.has("msgpack")) {
retval = DataFormat::MSGPACK;
} else if (incoming) {
if (request.getContentType() == "application/json") {
retval = DataFormat::JSON;
} else if (request.getContentType() == "application/javascript") {
retval = DataFormat::JSONP;
} else if (request.getContentType() == "application/yaml") {
retval = DataFormat::YAML;
} else if (request.getContentType() == "application/xml") {
retval = DataFormat::XMLPLIST;
} else if (request.getContentType() == "application/msgpack") {
retval = DataFormat::MSGPACK;
} else if (request.getContentType() == "application/bundlehdr") {
retval = DataFormat::BUNDLEHDR;
}
} else { // outgoing
if (request.hasToken("Accept", "application/json")) {
retval = DataFormat::JSON;
} else if (request.hasToken("Accept", "application/javascript")) {
retval = DataFormat::JSONP;
} else if (request.hasToken("Accept", "application/yaml")) {
retval = DataFormat::YAML;
} else if (request.hasToken("Accept", "application/xml")) {
retval = DataFormat::XMLPLIST;
} else if (request.hasToken("Accept", "application/msgpack")) {
retval = DataFormat::MSGPACK;
} else if (request.hasToken("Accept", "application/bundlehdr")) {
retval = DataFormat::BUNDLEHDR;
}
}
return retval;
}
std::string encodeVariant( const libvariant::Variant& msg,
const Poco::Net::HTTPServerRequest& request,
ResponseDataFormat format = ResponseDataFormat::AUTO_GUESS ) {
if (format == ResponseDataFormat::AUTO_GUESS) {
format = getResponseDataFormat(request);
DataFormat format = DataFormat::AUTO_GUESS ) {
if (format == DataFormat::AUTO_GUESS) {
format = getDataFormat(request);
}
switch (format) {
case ResponseDataFormat::JSON:
case DataFormat::JSON:
return libvariant::SerializeJSON(msg, true);
case ResponseDataFormat::JSONP:
case DataFormat::JSONP:
{
Poco::Net::HTMLForm form(request);
std::string padding;
std::string padding("callback");
if (form.has("callback")) {
padding = form.get("callback");
} else if (form.has("jsonp")) {
} else if (form.has("jsonp") && !form.get("jsonp").empty()) {
padding = form.get("jsonp");
}
if (padding.empty()) {
@@ -349,13 +399,13 @@ protected:
ss << padding << "(" << libvariant::SerializeJSON(msg, false) << ")";
return ss.str();
}
case ResponseDataFormat::YAML:
case DataFormat::YAML:
return libvariant::SerializeYAML(msg);
case ResponseDataFormat::XMLPLIST:
case DataFormat::XMLPLIST:
return libvariant::SerializeXMLPLIST(msg, true);
case ResponseDataFormat::BUNDLEHDR:
case DataFormat::BUNDLEHDR:
return libvariant::SerializeBundleHdr(msg);
case ResponseDataFormat::MSGPACK:
case DataFormat::MSGPACK:
return libvariant::Serialize(msg, libvariant::SERIALIZE_MSGPACK);
default:
throw WebResourceHandlerException("Internal Server Error",
@@ -363,80 +413,122 @@ protected:
}
}
void handleResponse( const Poco::Net::HTTPServerRequest& request,
Poco::Net::HTTPServerResponse &resp,
const libvariant::Variant& msg,
const std::string& mime = "text/plain",
Poco::Net::HTTPResponse::HTTPStatus status = Poco::Net::HTTPResponse::HTTPStatus::HTTP_OK ) {
std::string getContentTypeString(DataFormat format) {
if (format == DataFormat::JSON) {
return "application/json";
} else if (format == DataFormat::JSONP) {
return "application/javascript";
} else if (format == DataFormat::YAML) {
return "application/yaml";
} else if (format == DataFormat::XMLPLIST) {
return "application/xml";
} else {
return "";
}
}
void setContentType(Poco::Net::HTTPServerResponse &resp, DataFormat format) {
auto type = getContentTypeString(format);
if (!type.empty()) {
resp.setContentType(type);
}
}
void handleResponse( Poco::Net::HTTPServerRequest& request,
Poco::Net::HTTPServerResponse &resp,
const libvariant::Variant& msg,
Poco::Net::HTTPResponse::HTTPStatus status = Poco::Net::HTTPResponse::HTTPStatus::HTTP_OK,
const std::string& mime = "" ) {
resp.setStatusAndReason(status);
auto format = getDataFormat(request);
if (mime.empty()) {
setContentType(resp, format);
} else {
resp.setContentType(mime);
}
std::stringstream ss;
ss << encodeVariant(msg, request, format);
std::string buffer = ss.str();
resp.sendBuffer(buffer.c_str(), buffer.size());
}
void handleResponse( Poco::Net::HTTPServerRequest& request,
Poco::Net::HTTPServerResponse &resp,
const std::string& msg,
Poco::Net::HTTPResponse::HTTPStatus status = Poco::Net::HTTPResponse::HTTPStatus::HTTP_OK,
const std::string& mime = "text/plain" ) {
resp.setStatusAndReason(status);
resp.setContentType(mime);
auto format = getResponseDataFormat(request);
resp.send() << encodeVariant(msg, request, format);
resp.sendBuffer(msg.c_str(), msg.size());
}
void handleResponse( const Poco::Net::HTTPServerRequest& request,
void handleErrorResponse( Poco::Net::HTTPServerRequest& request,
Poco::Net::HTTPServerResponse &resp,
const std::string& msg,
const std::string& mime = "text/plain",
Poco::Net::HTTPResponse::HTTPStatus status = Poco::Net::HTTPResponse::HTTPStatus::HTTP_OK ) {
Poco::Net::HTTPResponse::HTTPStatus status = Poco::Net::HTTPResponse::HTTPStatus::HTTP_INTERNAL_SERVER_ERROR,
const std::string& mime = "text/plain" ) {
resp.setStatusAndReason(status);
resp.setContentType(mime);
resp.send() << msg;
std::stringstream ss;
ss << "ERROR: " << msg;
addDebugInfo(ss, request, resp);
std::string buffer = ss.str();
resp.sendBuffer(buffer.c_str(), buffer.size());
}
void handleErrorResponse( const Poco::Net::HTTPServerRequest& request,
Poco::Net::HTTPServerResponse &resp,
const std::string& msg,
Poco::Net::HTTPResponse::HTTPStatus status = Poco::Net::HTTPResponse::HTTPStatus::HTTP_INTERNAL_SERVER_ERROR ) {
resp.setStatusAndReason(status);
resp.setContentType("text/plain");
std::ostream& out = resp.send();
out << "ERROR: " << msg;
addDebugInfo(out, request, resp);
}
void handleErrorResponse( const Poco::Net::HTTPServerRequest& request,
void handleErrorResponse( Poco::Net::HTTPServerRequest& request,
Poco::Net::HTTPServerResponse &resp,
Poco::Net::HTTPResponse::HTTPStatus status = Poco::Net::HTTPResponse::HTTPStatus::HTTP_INTERNAL_SERVER_ERROR ) {
resp.setStatusAndReason(status);
resp.setContentType("text/plain");
std::ostream& out = resp.send();
out << "ERROR: " << resp.getReason();
addDebugInfo(out, request, resp);
std::stringstream ss;
ss << "ERROR: " << resp.getReason();
addDebugInfo(ss, request, resp);
std::string buffer = ss.str();
resp.sendBuffer(buffer.c_str(), buffer.size());
}
void handleErrorResponse( const Poco::Net::HTTPServerRequest& request,
void handleErrorResponse( Poco::Net::HTTPServerRequest& request,
Poco::Net::HTTPServerResponse &resp,
const WebResourceHandlerException& e ) {
resp.setStatusAndReason(e.getStatusCode());
resp.setContentType("text/plain");
std::ostream& out = resp.send();
out << "ERROR: " << e.what();
addDebugInfo(out, request, resp);
std::stringstream ss;
ss << "ERROR: " << e.what();
addDebugInfo(ss, request, resp);
std::string buffer = ss.str();
resp.sendBuffer(buffer.c_str(), buffer.size());
}
void handleErrorResponse( const Poco::Net::HTTPServerRequest& request,
void handleErrorResponse( Poco::Net::HTTPServerRequest& request,
Poco::Net::HTTPServerResponse &resp,
const std::exception& e,
Poco::Net::HTTPResponse::HTTPStatus status = Poco::Net::HTTPResponse::HTTPStatus::HTTP_INTERNAL_SERVER_ERROR ) {
resp.setStatusAndReason(status);
resp.setContentType("text/plain");
std::ostream& out = resp.send();
out << "ERROR: " << e.what();
addDebugInfo(out, request, resp);
std::stringstream ss;
ss << "ERROR: " << e.what();
addDebugInfo(ss, request, resp);
std::string buffer = ss.str();
resp.sendBuffer(buffer.c_str(), buffer.size());
}
void handleRequireAuthenticationResponse( const Poco::Net::HTTPServerRequest& request,
Poco::Net::HTTPServerResponse &resp,
const std::string& realm,
const std::string& body = std::string(),
Poco::Net::HTTPResponse::HTTPStatus status = Poco::Net::HTTPResponse::HTTPStatus::HTTP_UNAUTHORIZED ) {
void handleRequireAuthenticationResponse( Poco::Net::HTTPServerRequest& request,
Poco::Net::HTTPServerResponse &resp,
const std::string& realm,
const std::string& body = "",
Poco::Net::HTTPResponse::HTTPStatus status = Poco::Net::HTTPResponse::HTTPStatus::HTTP_UNAUTHORIZED,
const std::string& mime = "text/plain" ) {
resp.requireAuthentication(realm);
resp.setContentType(mime);
std::stringstream ss;
if (body.empty()) {
resp.send() << "ERROR: " << resp.getReason();
ss << "ERROR: " << resp.getReason();
} else {
resp.send() << body;
ss << body;
}
std::string buffer = ss.str();
resp.sendBuffer(buffer.c_str(), buffer.size());
}
void addDebugInfo(std::ostream& out, const Poco::Net::HTTPServerRequest& req, Poco::Net::HTTPServerResponse &resp) {
@@ -500,6 +592,107 @@ protected:
}
}
template<class EventQueue, class HandlerFactory>
void __handleEventWebSocket(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp, EventQueue* eventQueue, HandlerFactory* factory) {
try {
Poco::Net::WebSocket ws(req, resp);
std::cout << "WebSocket connection established." << std::endl;
bool breakLoop = false;
factory->getIOService().post([&]() {
char buffer[1024];
int flags;
int n;
bool timeout = false;
do {
try {
timeout = false;
n = ws.receiveFrame(buffer, sizeof(buffer), flags);
// ToDo: Handle ping and other shit.
std::cout << Poco::format("Frame received (length=%d, flags=0x%x).", n, unsigned(flags)) << std::endl;
} catch (typename EventQueue::TimeoutException& e) {
// Timeout may happen, nothing fatal.
timeout = true;
} catch (Poco::Net::NetException& e) {
breakLoop= true;
ROS_ERROR_STREAM("Exception in WebSocket receive activity: " << e.what());
} catch (std::exception& e) {
ROS_ERROR_STREAM("WebServer Receive Exception caught: " << e.what());
}
} while (timeout || (!breakLoop && n > 0 && (flags & Poco::Net::WebSocket::FRAME_OP_BITMASK) != Poco::Net::WebSocket::FRAME_OP_CLOSE));
breakLoop= true;
eventQueue->cancelPopEvent();
});
while (!breakLoop) {
libvariant::Variant msg;
msg = eventQueue->popEvent();
std::string encoded = encodeVariant(msg, req);
ws.sendFrame(encoded.c_str(), encoded.size(), Poco::Net::WebSocket::FRAME_FLAG_FIN | Poco::Net::WebSocket::FRAME_OP_TEXT);
}
std::cout << "WebSocket connection closed." << std::endl;
} catch (Poco::Net::WebSocketException& exc) {
std::cout << "WebSocketException caught: " << exc.what() << std::endl;
switch (exc.code()) {
case Poco::Net::WebSocket::WS_ERR_HANDSHAKE_UNSUPPORTED_VERSION:
resp.set("Sec-WebSocket-Version", Poco::Net::WebSocket::WEBSOCKET_VERSION);
// fallthrough
case Poco::Net::WebSocket::WS_ERR_NO_HANDSHAKE:
case Poco::Net::WebSocket::WS_ERR_HANDSHAKE_NO_VERSION:
case Poco::Net::WebSocket::WS_ERR_HANDSHAKE_NO_KEY:
resp.setStatusAndReason(Poco::Net::HTTPResponse::HTTP_BAD_REQUEST);
resp.setContentLength(0);
resp.send();
break;
}
} catch (std::exception& e) {
ROS_ERROR_STREAM("WebServer Send Exception caught: " << e.what());
}
}
template<class EventQueue, class HandlerFactory>
void __handleEventLongPolling(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp, EventQueue* eventQueue, HandlerFactory* factory) {
libvariant::Variant msg;
bool breakLoop = false;
boost::asio::deadline_timer connCheckTimer(factory->getIOService());
Poco::Net::HTTPServerRequestImpl* reqImpl = static_cast<Poco::Net::HTTPServerRequestImpl*>(&req);
std::mutex _m;
std::function<void(const boost::system::error_code& error)> connCheckFun = [&](const boost::system::error_code& error) {
if (error != boost::asio::error::operation_aborted) {
std::lock_guard<std::mutex> _l(_m);
// How to find out whether the client disconnected: Monitor the socket with Posix "select":
// If select says that there is data, available() returns 0 then apparently the client has disconnected.
// WTF, is this really the only way to notice that a client properly disconnected? Internet says so.
auto sockfs = reqImpl->socket().impl()->sockfd();
fd_set fds;
FD_ZERO(&fds);
FD_SET(sockfs, &fds);
struct timeval timeout = {0, 1000};
int sel = select(sockfs + 1, &fds, 0, 0, &timeout);
if (sel < 0 || (sel > 0 && reqImpl->socket().available() == 0)) {
ROS_ERROR_STREAM("Apparently we have a disconnect in device event queue \"" << eventQueue->clientID << "\"");
breakLoop = true;
eventQueue->cancelPopEvent();
}
connCheckTimer.expires_from_now(boost::posix_time::milliseconds(100));
connCheckTimer.async_wait(connCheckFun);
}
};
connCheckTimer.expires_from_now(boost::posix_time::milliseconds(100));
connCheckTimer.async_wait(connCheckFun);
try {
msg = eventQueue->popEvent();
} catch (typename EventQueue::TimeoutException& e) {
msg["eventtype"] = "timeoutevent";
}
{
std::lock_guard<std::mutex> _l(_m);
breakLoop = true;
connCheckTimer.cancel();
}
handleResponse(req, resp, msg);
}
};
@@ -138,6 +138,7 @@ void DirectoryResourceHandler::serveRegularFile(Poco::Net::HTTPServerRequest &re
if (type != parent->contentTypes.end()) {
resp.setContentType(type->second);
}
resp.setContentLength(boost::filesystem::file_size(file));
filesystem::ifstream fIn;
fIn.open(file, std::ios::in);
std::ostream& out = resp.send();
@@ -0,0 +1,65 @@
#pragma once
#include <map>
#include <functional>
namespace hos {
namespace utils {
template<class Key, class RetType, class... Args>
class FunctionTable {
public:
typedef std::function<RetType(Args...)> Func_t;
typedef std::function<Func_t(const Key&)> MissingKeyFunc_t;
protected:
std::map<Key, Func_t> _funcs;
MissingKeyFunc_t missingKeyFunc;
public:
FunctionTable(MissingKeyFunc_t missingKeyFunc = [](const Key&)->Func_t {
return [](Args...)->RetType { return RetType(); };
}) : missingKeyFunc(missingKeyFunc) {}
void add(const Key& key, const std::function<RetType(Args...)>& func) {
_funcs.insert({key, func});
}
std::function<RetType(Args...)> operator[](const Key& key) {
auto f = _funcs.find(key);
if (f != _funcs.end()) {
return f->second;
} else {
return missingKeyFunc(key);
}
}
RetType call(const Key& key, Args... args) {
return (*this)[key](args...);
}
void forEach(std::function<void(const Key&, const RetType&)> collector, Args... args) {
for (auto f : _funcs) {
collector(f.first, f.second(args...));
}
}
/*typename std::map<Key, Func_t>::iterator begin() {
return _funcs.begin();
}
typename std::map<Key, Func_t>::iterator end() {
return _funcs.end();
}*/
};
} // namespace utils
} // namespace hos
+7 -2
View File
@@ -1,11 +1,16 @@
if (CMAKE_BUILD_TYPE STREQUAL "Debug")
set(POCO_LIB_POSTFIX "d")
else (CMAKE_BUILD_TYPE STREQUAL "Debug")
set(POCO_LIB_POSTFIX "")
endif (CMAKE_BUILD_TYPE STREQUAL "Debug")
find_path(Poco_INCLUDE_DIR Poco/SharedLibrary.h
PATHS $ENV{HOS_PREFIX}/poco
PATH_SUFFIXES include
)
find_library(Poco_FOUNDATION_LIB PocoFoundation
find_library(Poco_FOUNDATION_LIB PocoFoundation${POCO_LIB_POSTFIX}
PATHS $ENV{HOS_PREFIX}/poco
PATH_SUFFIXES lib lib64
)
@@ -18,7 +23,7 @@ SET(Comp_List "")
IF(Poco_FOUNDATION_LIB)
SET(Comp_List "Foundation")
FOREACH(COMPONENT ${Poco_FIND_COMPONENTS})
FIND_LIBRARY(LIB${COMPONENT} "Poco${COMPONENT}" PATHS ${Poco_LIBRARY_DIR})
FIND_LIBRARY(LIB${COMPONENT} "Poco${COMPONENT}${POCO_LIB_POSTFIX}" PATHS ${Poco_LIBRARY_DIR})
IF (LIB${COMPONENT})
LIST(APPEND Poco_LIBRARIES "${LIB${COMPONENT}}")
LIST(APPEND Comp_List "${COMPONENT}")